Merge pull request #94 from strausmann/feat/phase-2-job-persistence

feat(jobs): Phase 2 Job Persistence (DB-backed + Restart Recovery + Cleanup)

Author: strausmann

backend/app/api/routes/batches.py

@@ -0,0 +1,69 @@
+"""Phase 2: GET /api/batches/{id} — Snapshot für Hangar Result-Page Initial-Render."""
+
+from __future__ import annotations
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.dependencies import AuthContext
+from app.auth.scope_deps import require_read
+from app.db.session import get_session
+from app.models.job import JobState
+from app.repositories import jobs as jobs_repo
+from app.repositories import print_batches as batches_repo
+from app.schemas.batch_read import BatchRead, BatchSummary
+from app.schemas.job import JobRead
+
+router = APIRouter(prefix="/api/batches", tags=["batches"])
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]
+ReadAuthDep = Annotated[AuthContext, Depends(require_read)]
+
+
+@router.get("/{batch_id}", response_model=BatchRead)
+async def get_batch(
+    batch_id: UUID,
+    session: SessionDep,
+    _auth: ReadAuthDep,
+) -> BatchRead:
+    """Snapshot eines Batches + aller aktuellen Job-States.
+
+    Wird von Hangar's /admin/print/result/{batch_id} für das initiale
+    Rendering genutzt. summary.all_terminal == False bedeutet, dass Hangar
+    einen SSE-Stream zu /api/events?batch_id=... öffnen sollte für
+    Live-Updates.
+    """
+    batch = await batches_repo.get(session, batch_id)
+    if batch is None:
+        raise HTTPException(status_code=404, detail="Batch not found")
+
+    # batch.job_ids is list[str] — convert to UUID for repo query
+    job_uuids = [UUID(jid) for jid in batch.job_ids]
+    fetched_jobs = await jobs_repo.list_by_ids(session, job_uuids)
+    job_map = {str(j.id): j for j in fetched_jobs}
+
+    # Reihenfolge entspricht batch.job_ids; cleanup-evicted Jobs werden übersprungen
+    ordered = [job_map[jid] for jid in batch.job_ids if jid in job_map]
+
+    summary = BatchSummary(
+        total=len(ordered),
+        queued=sum(1 for j in ordered if j.state == JobState.QUEUED.value),
+        printing=sum(1 for j in ordered if j.state == JobState.PRINTING.value),
+        done=sum(1 for j in ordered if j.state == JobState.DONE.value),
+        failed=sum(
+            1 for j in ordered if j.state in (JobState.FAILED.value, JobState.FAILED_RESTART.value)
+        ),
+        cancelled=sum(1 for j in ordered if j.state == JobState.CANCELLED.value),
+    )
+
+    return BatchRead(
+        id=batch.id,
+        printer_id=batch.printer_id,
+        created_by=batch.created_by,
+        created_at=batch.created_at,
+        jobs=[JobRead.model_validate(j) for j in ordered],
+        summary=summary,
+    )

backend/app/api/routes/print.py

@@ -79,7 +79,8 @@ async def create_print_job(
                 "loaded_mm": exc.loaded_mm,
             }
         return JSONResponse(status_code=http_status, content=body)
-    return PrintJobResponse(job_id=job_id, status="queued")
+    # Phase 2: submit_print_job gibt jetzt UUID zurück; Response-Schema erwartet str.
+    return PrintJobResponse(job_id=str(job_id), status="queued")
 
 
 @router.get(

backend/app/config.py

@@ -106,6 +106,16 @@ class Settings(BaseSettings):
     # Set to False during transition to avoid surprising existing automation.
     pangolin_bypass_scope_downgrade: bool = False
 
+    # Phase 2: Job-Retention für CleanupTask
+    job_retention_days: int = Field(
+        default=30,
+        ge=1,
+        description=(
+            "Terminal Jobs (DONE/FAILED/FAILED_RESTART/CANCELLED) werden nach diesem Zeitraum "
+            "vom CleanupTask gelöscht"
+        ),
+    )
+
     @field_validator("webhook_api_key")
     @classmethod
     def validate_api_key_length(cls, v: SecretStr) -> SecretStr:

backend/app/main.py

@@ -76,6 +76,7 @@
 from app import __version__
 from app.api.error_handlers import register_error_handlers
 from app.api.routes import batch as batch_routes
+from app.api.routes import batches as batches_routes
 from app.api.routes import events as events_routes
 from app.api.routes import jobs as jobs_routes
 from app.api.routes import lookup as lookup_routes
@@ -91,20 +92,22 @@
 from app.db.engine import async_session, engine
 from app.db.lifespan import (
     ensure_printer_state,
-    recover_inflight_jobs,
     run_migrations,
     seed_templates,
     upsert_runtime_printer,
     verify_alembic_at_head,
 )
 from app.db.session import get_session
 from app.integrations.registry import IntegrationRegistry
+from app.models.printer import Printer as _Printer
 from app.printer_backends import BackendRegistry
 from app.printer_backends.exceptions import SnmpDiscoveryError
 from app.printer_backends.snmp_helper import query_model_pjl
 from app.printer_models.registry import ModelRegistry
 from app.schemas.readiness import ReadinessResponse
+from app.services.cleanup_task import CleanupTask
 from app.services.event_bus import EventBus
+from app.services.job_store_sqlite import SQLiteJobStore
 from app.services.label_renderer import LabelRenderer
 from app.services.lookup_service import AppLookupService
 from app.services.print_queue import PrintQueue
@@ -270,13 +273,25 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
     # 4. DB-bound init — plugin registry and template cache are populated.
     async with async_session() as s:
-        await recover_inflight_jobs(s)
+        # Phase 2: recover_inflight_jobs() entfernt (Spec R1-C1) —
+        # PrintQueue.start() übernimmt Recovery mit korrekter QUEUED/PRINTING-Differenzierung.
         await seed_templates(s, TemplateLoader)
         db_printer_id = await upsert_runtime_printer(s, settings)
         await ensure_printer_state(s)
         await s.commit()
     # -------------------------------------------------------------------------
 
+    # Phase 2: JobStore + CleanupTask
+    # 'async_session' ist die async_sessionmaker aus app.db.engine (R2-M5)
+    job_store = SQLiteJobStore(async_session)
+
+    cleanup_task = CleanupTask(
+        store=job_store,
+        retention_days=settings.job_retention_days,
+    )
+    await cleanup_task.start()
+    app.state.cleanup_task = cleanup_task
+
     discovery_host = settings.pt750w_host or ""
     if discovery_host and settings.printer_discover_via_snmp:
         model_id = await _resolve_model_id(settings, discovery_host)
@@ -296,15 +311,51 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     tape_registry = TapeRegistry()
     printer = driver.make_queue_printer(tape_registry, printer_id=db_printer_id)
 
+    if db_printer_id is None:
+        # Wenn kein Host konfiguriert ist (Mock-Backend / CI), liefert
+        # upsert_runtime_printer None zurück und fügt keine Printer-Row ein.
+        # make_queue_printer erzeugt dann eine neue uuid4. Damit
+        # jobs.printer_id (FK → printers.id) bei save_queued nicht verletzt
+        # wird, legen wir hier eine Stub-Row an. slug wird auf str(id) gesetzt
+        # (eindeutig durch UUID), damit der UNIQUE-Constraint nicht verletzt wird.
+        _stub_slug = str(printer.id)
+        async with async_session() as s:
+            # Defensive idempotency: in non-mock production paths the printer_id
+            # is explicit and would be reused, so the existing check matters
+            # there. In mock paths (printer_id=None), a fresh uuid4 means
+            # existing is always None.
+            existing = await s.get(_Printer, printer.id)
+            if existing is None:
+                s.add(
+                    _Printer(
+                        id=printer.id,
+                        name=f"stub-{printer.id}",
+                        slug=_stub_slug,
+                        model=model_id.lower(),
+                        backend=settings.printer_backend,
+                    )
+                )
+                await s.commit()
+
     # --- SSE EventBus ---
     event_bus = EventBus(queue_size=settings.sse_queue_size)
     app.state.event_bus = event_bus
     # ----- end SSE ------
 
+    # Shared LabelRenderer reused by both PrintService, preview endpoint and
+    # PrintQueue Recovery. Constructing it once avoids repeated font-loading
+    # overhead on every POST /api/render/preview request.
+    # Moved before PrintQueue construction so Recovery in queue.start() can use it.
+    shared_renderer = LabelRenderer()
+    app.state.label_renderer = shared_renderer
+
     pq_producer = PrintQueueProducer(bus=event_bus)
     queue = PrintQueue(
         printers=[printer],
         on_state_change=pq_producer.handle_transition,
+        store=job_store,
+        renderer=shared_renderer,
+        loader=TemplateLoader,
     )
     await queue.start()
 
@@ -329,18 +380,14 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     app.state.printer_id = printer.id
     app.state.printer_host = discovery_host
     app.state.printer_snmp_community = settings.printer_snmp_community
-    # Shared LabelRenderer reused by both PrintService and the preview endpoint.
-    # Constructing it once avoids repeated font-loading overhead on every
-    # POST /api/render/preview request.
-    shared_renderer = LabelRenderer()
-    app.state.label_renderer = shared_renderer
     app.state.print_service = PrintService(
         template_loader=TemplateLoader,
         renderer=shared_renderer,
         print_queue=queue,
         lookup_service=AppLookupService(),
         printer_id=printer.id,
         backend=backend,
+        store=job_store,
     )
 
     try:
@@ -349,6 +396,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         if status_producer is not None:
             await status_producer.stop()
         await queue.stop(timeout_s=settings.printer_queue_timeout_s)
+        await cleanup_task.stop()
         await engine.dispose()
         # Close shared HTTP clients held by integration plugins that support it.
         # Plugins that pre-date connection pooling may not have aclose(); skip them.
@@ -597,6 +645,7 @@ async def readiness(
     register_error_handlers(app)
     app.include_router(print_router)
     app.include_router(batch_routes.router)
+    app.include_router(batches_routes.router)
     app.include_router(events_routes.router)
     app.include_router(printers_routes.router)
     app.include_router(templates_routes.router)

backend/app/repositories/jobs.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import UUID
 
-from sqlalchemy import select, update
+from sqlalchemy import delete, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import col
 
@@ -160,12 +160,98 @@ async def mark_inflight_as_failed_restart(session: AsyncSession) -> int:
     return int(result.rowcount)  # type: ignore[attr-defined]  # rowcount on UPDATE result
 
 
-async def list_active(session: AsyncSession) -> list[Job]:
-    """Return all jobs in QUEUED or PRINTING state (covered by ix_jobs_state)."""
+async def list_active(
+    session: AsyncSession,
+    *,
+    printer_id: UUID | None = None,
+) -> list[Job]:
+    """Return all jobs in QUEUED or PRINTING state (covered by ix_jobs_state).
+
+    Phase 2: optional printer_id filter for PrintQueue.start() recovery.
+    """
     inflight = (JobState.QUEUED.value, JobState.PRINTING.value)
-    result = await session.execute(
-        select(Job)
-        .where(col(Job.state).in_(inflight))  # col() gives proper Column typing for .in_()
-        .order_by(col(Job.created_at))  # col() gives proper Column typing
+    stmt = select(Job).where(col(Job.state).in_(inflight))
+    if printer_id is not None:
+        stmt = stmt.where(col(Job.printer_id) == printer_id)
+    stmt = stmt.order_by(col(Job.created_at))
+    result = await session.execute(stmt)
+    return list(result.scalars())
+
+
+async def mark_printing_as_failed_restart(
+    session: AsyncSession,
+    printer_id: UUID,
+) -> int:
+    """Phase 2: UPDATE only PRINTING jobs for a specific printer to
+    FAILED_RESTART with error='printer_interrupted'.
+
+    Used at PrintQueue.start() — QUEUED jobs are NOT affected because
+    they will be re-enqueued cleanly. Only PRINTING jobs are ambiguous
+    (printer may have completed before crash but Hub couldn't update DB).
+
+    Returns the count of affected rows.
+    """
+    stmt = (
+        update(Job)
+        .where(
+            col(Job.printer_id) == printer_id,
+            col(Job.state) == JobState.PRINTING.value,
+        )
+        .values(
+            state=JobState.FAILED_RESTART.value,
+            error="printer_interrupted",
+            finished_at=datetime.now(UTC),
+        )
+        .execution_options(synchronize_session="fetch")
     )
+    result = await session.execute(stmt)
+    await session.commit()
+    return int(result.rowcount)  # type: ignore[attr-defined]
+
+
+async def list_by_ids(
+    session: AsyncSession,
+    job_ids: list[UUID],
+) -> list[Job]:
+    """Bulk-Fetch jobs by ids — order not guaranteed, caller re-orders.
+
+    Phase 2: used by GET /api/batches/{id} to load all jobs referenced
+    by a PrintBatch.job_ids list in a single SQL query.
+    """
+    if not job_ids:
+        return []
+    result = await session.execute(select(Job).where(col(Job.id).in_(job_ids)))
     return list(result.scalars())
+
+
+async def evict_terminal_older_than(
+    session: AsyncSession,
+    age: timedelta,
+) -> int:
+    """Phase 2 cleanup: DELETE terminal jobs older than age.
+
+    Terminal = DONE | FAILED | FAILED_RESTART | CANCELLED.
+    Comparison is on finished_at (set whenever a job leaves a non-terminal state).
+
+    Jobs with finished_at IS NULL are NOT deleted (NULL < cutoff is SQL UNKNOWN,
+    which is falsy in WHERE). This is intentional — protects pre-Phase-2 rows
+    that may not have finished_at set.
+
+    Returns the count of deleted rows.
+    """
+    terminal = (
+        JobState.DONE.value,
+        JobState.FAILED.value,
+        JobState.FAILED_RESTART.value,
+        JobState.CANCELLED.value,
+    )
+    cutoff = datetime.now(UTC) - age
+    stmt = (
+        delete(Job)
+        .where(col(Job.state).in_(terminal))
+        .where(col(Job.finished_at) < cutoff)
+        .execution_options(synchronize_session="fetch")
+    )
+    result = await session.execute(stmt)
+    await session.commit()
+    return int(result.rowcount)  # type: ignore[attr-defined]