bambuddy/backend/app/services/timelapse_jobs.py

"""Timelapse editor re-encodes, run as background jobs.

Re-encoding a long timelapse takes minutes. Run inside the request, a browser
or proxy that gave up left ffmpeg running with nobody waiting for the result.
A job runs on its own instead: the editor polls its progress, and cancelling
it stops ffmpeg. One job per archive at a time; the last one is kept so a
reopened editor can show how it ended.
"""

import asyncio
import contextlib
import logging
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"


@dataclass
class TimelapseJob:
    archive_id: int
    status: str = RUNNING
    # 0.0 to 1.0; None until ffmpeg reports its first position.
    progress: float | None = None
    output_path: str | None = None
    message: str | None = None
    error: str | None = None
    task: asyncio.Task | None = field(default=None, repr=False)

    def set_progress(self, progress: float) -> None:
        self.progress = min(max(progress, 0.0), 1.0)


class JobAlreadyRunning(Exception):
    pass


# Runs the re-encode; returns (output path relative to the data dir, message).
JobRunner = Callable[[TimelapseJob], Awaitable[tuple[str, str]]]

_jobs: dict[int, TimelapseJob] = {}


def get_job(archive_id: int) -> TimelapseJob | None:
    return _jobs.get(archive_id)


def start_job(archive_id: int, run: JobRunner) -> TimelapseJob:
    current = _jobs.get(archive_id)
    if current is not None and current.status == RUNNING:
        raise JobAlreadyRunning
    job = TimelapseJob(archive_id=archive_id)
    _jobs[archive_id] = job
    job.task = asyncio.create_task(_run(job, run))
    return job


async def _run(job: TimelapseJob, run: JobRunner) -> None:
    try:
        job.output_path, job.message = await run(job)
        job.progress = 1.0
        job.status = COMPLETED
    except asyncio.CancelledError:
        job.status = CANCELLED
    except Exception as e:
        logger.error("Timelapse processing failed for archive %s: %s", job.archive_id, e)
        job.error = str(e)
        job.status = FAILED


async def cancel_job(archive_id: int) -> bool:
    """Cancel the archive's running job; False when none is running."""
    job = _jobs.get(archive_id)
    if job is None or job.status != RUNNING or job.task is None:
        return False
    job.task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await job.task
    return True


async def cancel_all() -> None:
    for archive_id in list(_jobs):
        await cancel_job(archive_id)