feat(repo): jobs_repo Helper für Phase 2 JobStore

- mark_printing_as_failed_restart(printer_id) — nur PRINTING affected
- list_active(printer_id=None) — optionaler Filter
- evict_terminal_older_than(age) — Cleanup-Helper

Refs #93

Author: strausmann

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,82 @@ 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(
+    stmt = (
         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
+        .where(col(Job.state).in_(inflight))
+        .order_by(col(Job.created_at))
     )
+    if printer_id is not None:
+        stmt = stmt.where(col(Job.printer_id) == printer_id)
+    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 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).
+
+    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]

backend/tests/unit/repositories/__init__.py

@@ -0,0 +1,29 @@
+"""Fixtures für Unit-Tests der Repository-Schicht.
+
+Stellt eine per-Test in-memory SQLite DB mit db_session-Fixture bereit.
+FK-Enforcement bewusst NICHT aktiviert — Phase-2-Tests nutzen uuid4()
+als printer_id ohne echte Printer-Rows anzulegen (Unit-Scope).
+"""
+
+from __future__ import annotations
+
+import app.models  # noqa: F401 — registriert alle Models mit SQLModel.metadata
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlmodel import SQLModel
+
+
+@pytest_asyncio.fixture
+async def _engine():
+    eng = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with eng.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+    yield eng
+    await eng.dispose()
+
+
+@pytest_asyncio.fixture
+async def db_session(_engine):
+    factory = async_sessionmaker(_engine, expire_on_commit=False)
+    async with factory() as s:
+        yield s

backend/tests/unit/repositories/conftest.py

@@ -0,0 +1,87 @@
+"""Phase 2: neue jobs_repo Helper fuer JobStore Adapter."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from app.models.job import Job, JobState
+from app.repositories import jobs as jobs_repo
+
+
+@pytest.mark.asyncio
+async def test_mark_printing_as_failed_restart_only_printing(db_session):
+    """mark_printing_as_failed_restart darf QUEUED-Jobs NICHT aendern."""
+    printer_id = uuid4()
+    other_printer_id = uuid4()
+
+    queued = await jobs_repo.create_queued(
+        db_session, printer_id=printer_id,
+        template_key="t", payload={"k": "v"},
+    )
+    printing = await jobs_repo.create_queued(
+        db_session, printer_id=printer_id,
+        template_key="t", payload={"k": "v"},
+    )
+    await jobs_repo.mark_printing(db_session, printing.id)
+
+    other_printing = await jobs_repo.create_queued(
+        db_session, printer_id=other_printer_id,
+        template_key="t", payload={"k": "v"},
+    )
+    await jobs_repo.mark_printing(db_session, other_printing.id)
+
+    affected = await jobs_repo.mark_printing_as_failed_restart(
+        db_session, printer_id,
+    )
+    assert affected == 1  # nur das eine PRINTING auf unserem printer
+
+    await db_session.refresh(queued)
+    await db_session.refresh(printing)
+    await db_session.refresh(other_printing)
+
+    assert queued.state == JobState.QUEUED.value
+    assert printing.state == JobState.FAILED_RESTART.value
+    assert printing.error == "printer_interrupted"
+    assert printing.finished_at is not None
+    assert other_printing.state == JobState.PRINTING.value  # anderer printer unangetastet
+
+
+@pytest.mark.asyncio
+async def test_list_active_filterable_by_printer(db_session):
+    """list_active(printer_id=...) liefert nur Jobs des Druckers."""
+    p1, p2 = uuid4(), uuid4()
+    j1 = await jobs_repo.create_queued(db_session, printer_id=p1, template_key="t", payload={})
+    j2 = await jobs_repo.create_queued(db_session, printer_id=p2, template_key="t", payload={})
+
+    all_active = await jobs_repo.list_active(db_session)
+    assert {j.id for j in all_active} == {j1.id, j2.id}
+
+    p1_only = await jobs_repo.list_active(db_session, printer_id=p1)
+    assert {j.id for j in p1_only} == {j1.id}
+
+
+@pytest.mark.asyncio
+async def test_evict_terminal_older_than(db_session):
+    """evict loescht DONE/FAILED/CANCELLED/FAILED_RESTART aelter als age."""
+    printer_id = uuid4()
+    old_done = await jobs_repo.create_queued(db_session, printer_id=printer_id, template_key="t", payload={})
+    await jobs_repo.mark_printing(db_session, old_done.id)
+    await jobs_repo.mark_done(db_session, old_done.id, result={})
+    # backdate finished_at by hand for test
+    old_done.finished_at = datetime.now(UTC) - timedelta(days=35)
+    await db_session.commit()
+
+    young_done = await jobs_repo.create_queued(db_session, printer_id=printer_id, template_key="t", payload={})
+    await jobs_repo.mark_printing(db_session, young_done.id)
+    await jobs_repo.mark_done(db_session, young_done.id, result={})  # finished_at is now()
+
+    queued = await jobs_repo.create_queued(db_session, printer_id=printer_id, template_key="t", payload={})  # not terminal
+
+    deleted = await jobs_repo.evict_terminal_older_than(db_session, age=timedelta(days=30))
+    assert deleted == 1
+
+    assert await jobs_repo.get(db_session, old_done.id) is None
+    assert await jobs_repo.get(db_session, young_done.id) is not None
+    assert await jobs_repo.get(db_session, queued.id) is not None