alembic/app/services/pipeline_runner.py

import asyncio
import time
from pathlib import Path

from app.db import SessionLocal
from app.models import JobRun
from app.settings import settings

# Single global lock replacing /var/lock/sldl-pipeline.lock + flock -n. One
# process (this one) now owns every pipeline invocation, so one asyncio.Lock
# is enough -- deliberately not per-resource, matching the original flock's
# all-jobs-share-one-lock behavior exactly rather than over-engineering it.
_lock = asyncio.Lock()

# How long a SCHEDULED run waits for the lock before giving up (recorded as
# skipped_lock). The old flock -n instant-skip starves fixed-time schedules:
# the Sunday 08:30 strip-mb-tags run took 10m19s on 2026-07-12, so the 08:35
# and 08:40 jobs both hit the held lock and silently lost their only slot of
# the week. Scheduled jobs have nobody watching, so queueing (bounded) beats
# skipping; manual runs keep the instant skip because a person clicking "Run
# now" should get an immediate answer, not a silent 30-minute wait.
SCHEDULED_LOCK_WAIT_SECONDS = 1800.0

_ENV_PASSTHROUGH_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TZ")


async def _try_acquire_nowait() -> bool:
    """asyncio.Lock has no acquire_nowait(). asyncio.wait_for(lock.acquire(),
    timeout=0) looks like the obvious idiom but is broken in practice: it
    wraps acquire() in a Task, and the Task's first iteration and the
    timeout-0 callback are both scheduled via the event loop with no
    guaranteed ordering, so it times out almost every time even when the
    lock is completely uncontended (confirmed empirically -- the very first,
    uncontended call failed every time). The correct pattern relies on
    Lock.acquire()'s fast path never suspending when the lock is free: the
    .locked() check and the subsequent acquire() happen with no `await`
    point in between, so nothing else can interleave in this single-
    threaded event loop."""
    if _lock.locked():
        return False
    await _lock.acquire()
    return True


def _subprocess_env() -> dict:
    import os

    env = {k: v for k, v in os.environ.items() if k in _ENV_PASSTHROUGH_KEYS}
    env.setdefault("PATH", "/opt/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")
    env["ALEMBIC_CONFIG_DIR"] = str(settings.alembic_config_dir)
    env["MUSIC_DATA_DIR"] = str(settings.music_data_dir)
    env["PIPELINE_DIR"] = str(settings.pipeline_dir)
    env["BEETSDIR"] = str(settings.beets_dir)
    env["MIN_ARTIST_DIRS"] = str(settings.min_artist_dirs)
    return env


def _summarize_log(log_path: Path, max_len: int = 200) -> str:
    """Best-effort one-liner for the jobs list: the log's last non-empty
    line. Scripts here already end their own runs with a human-readable
    '=== ... done ===' / summary line (see pipeline-status.sh's convention),
    so this is usually meaningful without any per-script special-casing."""
    try:
        lines = [line for line in log_path.read_text(errors="replace").splitlines() if line.strip()]
        return lines[-1][:max_len] if lines else ""
    except OSError:
        return ""


def _record_run(job_key: str, started_at: float, triggered_by: str, **fields) -> JobRun:
    """Insert a job_runs row and return the refreshed object."""
    db = SessionLocal()
    try:
        run = JobRun(job_key=job_key, started_at=started_at, triggered_by=triggered_by, **fields)
        db.add(run)
        db.commit()
        db.refresh(run)
        return run
    finally:
        db.close()


def _finalize_run(run_id: int, status: str, exit_code: int | None, log_path: Path) -> JobRun:
    """Fill in the terminal fields on a previously-'running' job_runs row."""
    db = SessionLocal()
    try:
        run = db.get(JobRun, run_id)
        run.finished_at = time.time()
        run.status = status
        run.exit_code = exit_code
        run.summary = _summarize_log(log_path)
        db.commit()
        db.refresh(run)
        return run
    finally:
        db.close()


async def _execute(
    job_key: str,
    argv: list[str],
    triggered_by: str,
    timeout: float | None,
    use_lock: bool,
    capture: bool,
    lock_wait: float | None,
) -> tuple[JobRun, str]:
    """Shared core for run_job/run_job_capture: acquire the lock (unless
    use_lock=False), record the run, exec the subprocess, and finalize. When
    capture=True, stdout+stderr is captured as text and returned; otherwise it
    streams straight to the log file. Returns (JobRun, output_text) -- the text
    is "" in the non-capture and skipped-lock cases.

    lock_wait: how long to wait for a held lock before recording
    skipped_lock. None picks the policy default: SCHEDULED_LOCK_WAIT_SECONDS
    for triggered_by="schedule", instant skip for everything else."""
    started_at = time.time()
    if lock_wait is None:
        lock_wait = SCHEDULED_LOCK_WAIT_SECONDS if triggered_by == "schedule" else 0.0

    acquired = False
    if use_lock:
        if not await _try_acquire_nowait():
            if lock_wait <= 0:
                return _record_run(job_key, started_at, triggered_by, finished_at=started_at, status="skipped_lock"), ""
            try:
                await asyncio.wait_for(_lock.acquire(), timeout=lock_wait)
            except asyncio.TimeoutError:
                return _record_run(job_key, started_at, triggered_by, finished_at=time.time(), status="skipped_lock"), ""
        acquired = True

    try:
        log_dir = settings.logs_dir / job_key.replace(":", "_")
        log_dir.mkdir(parents=True, exist_ok=True)
        log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log"
        run_id = _record_run(job_key, started_at, triggered_by, status="running", log_path=str(log_path)).id

        output_text = ""
        exit_code: int | None
        try:
            if capture:
                proc = await asyncio.create_subprocess_exec(
                    *argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env=_subprocess_env(),
                )
                try:
                    stdout_bytes, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
                    exit_code = proc.returncode
                    output_text = stdout_bytes.decode(errors="replace")
                except asyncio.TimeoutError:
                    proc.kill()
                    await proc.wait()
                    exit_code = -1
                    output_text += f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n"
                status = "success" if exit_code == 0 else "failed"
                log_path.write_text(output_text)
            else:
                with open(log_path, "wb") as log_file:
                    proc = await asyncio.create_subprocess_exec(
                        *argv, stdout=log_file, stderr=asyncio.subprocess.STDOUT, env=_subprocess_env(),
                    )
                    try:
                        exit_code = await asyncio.wait_for(proc.wait(), timeout=timeout)
                    except asyncio.TimeoutError:
                        proc.kill()
                        await proc.wait()
                        exit_code = -1
                        with open(log_path, "a") as f:
                            f.write(f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n")
                status = "success" if exit_code == 0 else "failed"
        except Exception as exc:
            exit_code = None
            status = "failed"
            msg = f"\n[pipeline_runner] exception before/while running: {exc!r}\n"
            if capture:
                output_text += msg
                log_path.write_text(output_text)
            else:
                with open(log_path, "a") as f:
                    f.write(msg)

        run = _finalize_run(run_id, status, exit_code, log_path)
        return run, output_text
    finally:
        if acquired:
            _lock.release()


async def run_job(
    job_key: str,
    argv: list[str],
    triggered_by: str = "schedule",
    timeout: float | None = None,
    use_lock: bool = True,
    lock_wait: float | None = None,
) -> JobRun:
    """Run a pipeline command under the shared mutual-exclusion lock,
    recording one job_runs row start-to-finish. If the lock is already held,
    scheduled runs (triggered_by="schedule") wait up to
    SCHEDULED_LOCK_WAIT_SECONDS for it before recording skipped_lock -- see
    that constant for why -- while manual/other runs record skipped_lock
    immediately (the old flock -n behavior, visible in the UI instead of
    silently skipping). Pass lock_wait to override either way.

    use_lock=False runs the command WITHOUT taking the pipeline lock, for
    read-only jobs that never touch the library (e.g. the status report).
    Those must never be starved by a long-running write job, and running
    them concurrently is safe."""
    run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False, lock_wait=lock_wait)
    return run


async def run_job_capture(
    job_key: str,
    argv: list[str],
    triggered_by: str = "manual",
    timeout: float | None = None,
    use_lock: bool = True,
    lock_wait: float | None = None,
) -> tuple[JobRun, str]:
    """Like run_job(), but captures stdout+stderr as text and returns it
    alongside the JobRun, instead of only writing it to the log file --
    for callers that need to parse structured (JSON-lines) output, e.g.
    dedup_review_service's dry-run scan. The full output is still written
    to a log file afterward so job_runs.log_path works the same as any
    other job."""
    return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True, lock_wait=lock_wait)


async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun:
    script = settings.pipeline_dir / "bin" / "run-playlist.sh"
    argv = [str(script)]
    if no_m3u:
        argv.append("--no-m3u")
    argv.append(playlist_name)
    # 45 min SLDL_TIMEOUT (run-playlist.sh) + generous slack for the
    # tagging/beets-import/M3U steps that follow it in the same script.
    return await run_job(f"playlist:{playlist_name}", argv, triggered_by=triggered_by, timeout=3300)


async def run_import_track(args: list[str], triggered_by: str = "manual") -> JobRun:
    script = settings.pipeline_dir / "bin" / "import-track.sh"
    return await run_job("manual:import", [str(script)] + args, triggered_by=triggered_by)


async def run_lib_script(
    job_key: str,
    script_name: str,
    args: list[str] | None = None,
    triggered_by: str = "schedule",
    timeout: float | None = None,
    use_lock: bool = True,
) -> JobRun:
    script = settings.pipeline_dir / "lib" / script_name
    return await run_job(
        job_key,
        [str(script)] + (args or []),
        triggered_by=triggered_by,
        timeout=timeout,
        use_lock=use_lock,
    )


async def run_beet(job_key: str, args: list[str], triggered_by: str = "schedule") -> JobRun:
    return await run_job(job_key, ["beet"] + args, triggered_by=triggered_by)