Making long running operations asynchronous

app/services/jobs.py

@@ -0,0 +1,74 @@
+"""Lightweight in-process tracking for long-running admin operations.
+
+Jobs live in a module-level dict, so this only works correctly for a single
+OS process (the app's current deployment: one uvicorn worker, no multi-process
+or multi-replica setup — see ``entrypoint.sh`` / ``docker-compose.yml``). If
+the app is ever scaled to multiple workers/replicas, this needs to move to
+shared storage (a DB table, Redis) so a poll request landing on a different
+process can still see the job.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+_MAX_FINISHED_AGE_SECONDS = 3600
+
+
+@dataclass
+class Job:
+    id: str
+    kind: str
+    status: str = "running"  # running | done | error
+    message: str = "Working…"
+    result: object | None = None
+    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    finished_at: datetime | None = None
+
+
+_jobs: dict[str, Job] = {}
+
+
+def get(job_id: str) -> Job | None:
+    return _jobs.get(job_id)
+
+
+def _evict_stale() -> None:
+    now = datetime.now(UTC)
+    stale = [
+        job_id
+        for job_id, job in _jobs.items()
+        if job.finished_at
+        and (now - job.finished_at).total_seconds() > _MAX_FINISHED_AGE_SECONDS
+    ]
+    for job_id in stale:
+        del _jobs[job_id]
+
+
+def start(kind: str, run: Callable[[Job], Awaitable[None]]) -> Job:
+    """Create a job, launch ``run(job)`` in the background, and return it immediately.
+
+    ``run`` is responsible for updating ``job.status``/``job.message``/``job.result``
+    to reflect the outcome, including catching its own errors. Any exception that
+    escapes anyway is caught here as a last resort so the job doesn't hang forever
+    in "running" state.
+    """
+    _evict_stale()
+    job = Job(id=uuid.uuid4().hex, kind=kind)
+    _jobs[job.id] = job
+
+    async def _runner() -> None:
+        try:
+            await run(job)
+        except Exception as exc:
+            job.status = "error"
+            job.message = f"{kind} failed: {exc}"
+        finally:
+            job.finished_at = datetime.now(UTC)
+
+    asyncio.create_task(_runner())
+    return job

app/templates/admin/_job_progress.html

@@ -0,0 +1,7 @@
+<div id="{{ container_id }}"
+     hx-get="{{ poll_url }}"
+     hx-trigger="load delay:1s"
+     hx-swap="outerHTML"
+     class="{{ extra_class or '' }}">
+  <p class="muted">⏳ {{ job.message }}</p>
+</div>

app/templates/admin/_job_progress_inline.html

@@ -0,0 +1,5 @@
+<span id="{{ container_id }}"
+      hx-get="{{ poll_url }}"
+      hx-trigger="load delay:1s"
+      hx-swap="outerHTML"
+      class="muted">⏳ {{ job.message }}</span>

app/templates/admin/_scope_import_result.html

@@ -1,3 +1,4 @@
+<div id="{{ container_id or 'scope-import-result-body' }}">
 {% if error %}
 <div class="card">
   <p style="color:var(--color-danger,#c00)">{{ error }}</p>
@@ -39,3 +40,4 @@
   {% endif %}
 </div>
 {% endif %}
+</div>

app/web/admin.py

@@ -16,7 +16,7 @@ from fastapi import (
 from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.database import get_db
+from app.core.database import AsyncSessionLocal, get_db
 from app.crud import audit as audit_crud
 from app.crud import augmented_part as augmented_part_crud
 from app.crud import augmented_part_note as augmented_part_note_crud
@@ -44,6 +44,7 @@ from app.models.unc_thread import UncSeries
 from app.models.user import User
 from app.services import augmented_part_sync as augmented_part_sync_service
 from app.services import fit_matching as fit_matching_service
+from app.services import jobs
 from app.services import odata as odata_service
 from app.services import scope_import as scope_import_service
 from app.web.templates import templates
@@ -1469,11 +1470,21 @@ async def scope_import_page(
     )
 
 
+_SCOPE_IMPORT_STATUS_ID = "scope-import-status"
+
+
+def _scope_import_progress_ctx(job: jobs.Job) -> dict:
+    return {
+        "container_id": _SCOPE_IMPORT_STATUS_ID,
+        "poll_url": f"/admin/scope-import/jobs/{job.id}",
+        "job": job,
+    }
+
+
 @router.post("/scope-import", response_class=HTMLResponse)
 async def upload_scope_import(
     request: Request,
     file: UploadFile = File(...),
-    db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("augmented_parts:write")),
 ):
     filename = file.filename or ""
@@ -1481,33 +1492,74 @@ async def upload_scope_import(
         return templates.TemplateResponse(
             request,
             "admin/_scope_import_result.html",
-            {"error": "Expected an .xlsx spreadsheet."},
+            {
+                "error": "Expected an .xlsx spreadsheet.",
+                "container_id": _SCOPE_IMPORT_STATUS_ID,
+            },
         )
     data = await file.read()
-    try:
-        result = await scope_import_service.import_from_bytes(db, data)
-    except Exception as exc:  # noqa: BLE001
+    actor_id = actor.id
+
+    async def _run(job: jobs.Job) -> None:
+        async with AsyncSessionLocal() as db:
+            try:
+                result = await scope_import_service.import_from_bytes(db, data)
+            except Exception as exc:  # noqa: BLE001
+                job.status = "error"
+                job.message = f"Could not read spreadsheet: {exc}"
+                return
+            actor_row = await user_crud.get(db, actor_id)
+            await audit_crud.log(
+                db,
+                actor=actor_row,
+                action="update",
+                resource_type="AugmentedPart",
+                resource_id=None,
+                description=(
+                    f"Scope import ({filename}): {result.created} created, "
+                    f"{result.updated} updated, {result.skipped} skipped, "
+                    f"{result.errors} errors"
+                ),
+            )
+            job.status = "done"
+            job.result = result
+
+    job = jobs.start("scope_import", _run)
+    return templates.TemplateResponse(
+        request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
+    )
+
+
+@router.get("/scope-import/jobs/{job_id}", response_class=HTMLResponse)
+async def scope_import_job_status(
+    request: Request,
+    job_id: str,
+    _: User = Depends(web_require("augmented_parts:write")),
+):
+    job = jobs.get(job_id)
+    if job is None:
         return templates.TemplateResponse(
             request,
             "admin/_scope_import_result.html",
-            {"error": f"Could not read spreadsheet: {exc}"},
+            {
+                "error": "Import job not found — it may have expired.",
+                "container_id": _SCOPE_IMPORT_STATUS_ID,
+            },
+        )
+    if job.status == "running":
+        return templates.TemplateResponse(
+            request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
+        )
+    if job.status == "error":
+        return templates.TemplateResponse(
+            request,
+            "admin/_scope_import_result.html",
+            {"error": job.message, "container_id": _SCOPE_IMPORT_STATUS_ID},
         )
-    await audit_crud.log(
-        db,
-        actor=actor,
-        action="update",
-        resource_type="AugmentedPart",
-        resource_id=None,
-        description=(
-            f"Scope import ({filename}): {result.created} created, "
-            f"{result.updated} updated, {result.skipped} skipped, "
-            f"{result.errors} errors"
-        ),
-    )
     return templates.TemplateResponse(
         request,
         "admin/_scope_import_result.html",
-        {"result": result},
+        {"result": job.result, "container_id": _SCOPE_IMPORT_STATUS_ID},
     )
 
 
@@ -1645,38 +1697,89 @@ async def create_augmented_part(
     )
 
 
+_AUGMENTED_PARTS_SYNC_TARGET_ID = "augmented-part-list"
+
+
 @router.post("/augmented-parts/sync", response_class=HTMLResponse)
 async def sync_augmented_parts(
     request: Request,
-    db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("augmented_parts:write")),
 ):
-    try:
-        result = await augmented_part_sync_service.sync(db)
-        await audit_crud.log(
-            db,
-            actor=actor,
-            action="update",
-            resource_type="AugmentedPart",
-            resource_id=None,
-            description=(
-                f"Sync: {result.created} created, {result.deleted} deleted, "
-                f"{result.skipped} skipped (no type match)"
-            ),
-        )
-        sync_result = (
-            f"Sync complete: {result.created} created, {result.deleted} deleted"
-            + (
-                f", {result.skipped} skipped (type not recognised)"
-                if result.skipped
-                else ""
+    actor_id = actor.id
+
+    async def _run(job: jobs.Job) -> None:
+        async with AsyncSessionLocal() as db:
+            try:
+                result = await augmented_part_sync_service.sync(db)
+            except Exception as exc:
+                job.status = "error"
+                job.message = f"Sync failed: {exc}"
+                return
+            actor_row = await user_crud.get(db, actor_id)
+            await audit_crud.log(
+                db,
+                actor=actor_row,
+                action="update",
+                resource_type="AugmentedPart",
+                resource_id=None,
+                description=(
+                    f"Sync: {result.created} created, {result.deleted} deleted, "
+                    f"{result.skipped} skipped (no type match)"
+                ),
             )
-            + "."
+            job.status = "done"
+            job.message = (
+                f"Sync complete: {result.created} created, {result.deleted} deleted"
+                + (
+                    f", {result.skipped} skipped (type not recognised)"
+                    if result.skipped
+                    else ""
+                )
+                + "."
+            )
+
+    job = jobs.start("augmented_part_sync", _run)
+    return templates.TemplateResponse(
+        request,
+        "admin/_job_progress.html",
+        {
+            "container_id": _AUGMENTED_PARTS_SYNC_TARGET_ID,
+            "poll_url": f"/admin/augmented-parts/sync/jobs/{job.id}",
+            "job": job,
+        },
+    )
+
+
+@router.get("/augmented-parts/sync/jobs/{job_id}", response_class=HTMLResponse)
+async def augmented_parts_sync_job_status(
+    request: Request,
+    job_id: str,
+    db: AsyncSession = Depends(get_db),
+    _: User = Depends(web_require("augmented_parts:write")),
+):
+    job = jobs.get(job_id)
+    if job is None:
+        form_ctx = await _augmented_part_form_ctx(db)
+        return templates.TemplateResponse(
+            request,
+            "admin/_augmented_parts_table.html",
+            {
+                **form_ctx,
+                **await _default_augmented_parts_table_ctx(db),
+                "sync_result": None,
+                "sync_error": "Sync job not found — it may have expired.",
+            },
+        )
+    if job.status == "running":
+        return templates.TemplateResponse(
+            request,
+            "admin/_job_progress.html",
+            {
+                "container_id": _AUGMENTED_PARTS_SYNC_TARGET_ID,
+                "poll_url": f"/admin/augmented-parts/sync/jobs/{job_id}",
+                "job": job,
+            },
         )
-        sync_error = None
-    except Exception as exc:
-        sync_result = None
-        sync_error = f"Sync failed: {exc}"
     form_ctx = await _augmented_part_form_ctx(db)
     return templates.TemplateResponse(
         request,
@@ -1684,8 +1787,8 @@ async def sync_augmented_parts(
         {
             **form_ctx,
             **await _default_augmented_parts_table_ctx(db),
-            "sync_result": sync_result,
-            "sync_error": sync_error,
+            "sync_result": job.message if job.status == "done" else None,
+            "sync_error": job.message if job.status == "error" else None,
         },
     )
 
@@ -2355,32 +2458,94 @@ async def augmented_part_fits(
     return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
 
 
+def _augmented_part_fits_target_id(augmented_part_id: int) -> str:
+    return f"ap-{augmented_part_id}-fits"
+
+
 @router.post(
     "/augmented-parts/{augmented_part_id}/fits/recompute", response_class=HTMLResponse
 )
 async def recompute_augmented_part_fits(
     request: Request,
     augmented_part_id: int,
+    actor: User = Depends(web_require("fits:write")),
+):
+    actor_id = actor.id
+
+    async def _run(job: jobs.Job) -> None:
+        async with AsyncSessionLocal() as db:
+            result = await fit_matching_service.recompute_for_part(
+                db, augmented_part_id
+            )
+            actor_row = await user_crud.get(db, actor_id)
+            await audit_crud.log(
+                db,
+                actor=actor_row,
+                action="update",
+                resource_type="Fit",
+                resource_id=augmented_part_id,
+                description=(
+                    f"Recomputed fits for AugmentedPart #{augmented_part_id}: "
+                    f"{result.created} created, {result.removed} removed"
+                ),
+            )
+            job.status = "done"
+            job.message = f"{result.created} created, {result.removed} removed."
+
+    job = jobs.start("fit_recompute_part", _run)
+    return templates.TemplateResponse(
+        request,
+        "admin/_job_progress.html",
+        {
+            "container_id": _augmented_part_fits_target_id(augmented_part_id),
+            "poll_url": f"/admin/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job.id}",
+            "job": job,
+            "extra_class": "ap-inline-section",
+        },
+    )
+
+
+@router.get(
+    "/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job_id}",
+    response_class=HTMLResponse,
+)
+async def augmented_part_fits_recompute_job_status(
+    request: Request,
+    augmented_part_id: int,
+    job_id: str,
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("fits:write")),
 ):
-    result = await fit_matching_service.recompute_for_part(db, augmented_part_id)
-    await audit_crud.log(
-        db,
-        actor=actor,
-        action="update",
-        resource_type="Fit",
-        resource_id=augmented_part_id,
-        description=(
-            f"Recomputed fits for AugmentedPart #{augmented_part_id}: "
-            f"{result.created} created, {result.removed} removed"
-        ),
-    )
+    job = jobs.get(job_id)
+    if job is None:
+        ctx = await _augmented_part_fits_ctx(
+            db,
+            augmented_part_id,
+            can_write=_can_write_fits(actor),
+            error="Recompute job not found — it may have expired.",
+        )
+        return templates.TemplateResponse(
+            request, "admin/_augmented_part_fits.html", ctx
+        )
+    if job.status == "running":
+        return templates.TemplateResponse(
+            request,
+            "admin/_job_progress.html",
+            {
+                "container_id": _augmented_part_fits_target_id(augmented_part_id),
+                "poll_url": (
+                    f"/admin/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job_id}"
+                ),
+                "job": job,
+                "extra_class": "ap-inline-section",
+            },
+        )
     ctx = await _augmented_part_fits_ctx(
         db,
         augmented_part_id,
         can_write=_can_write_fits(actor),
-        message=f"{result.created} created, {result.removed} removed.",
+        message=job.message if job.status == "done" else None,
+        error=job.message if job.status == "error" else None,
     )
     return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
 
@@ -2588,29 +2753,73 @@ async def delete_augmented_part_fit(
     return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
 
 
+_ALL_FITS_RECOMPUTE_STATUS_ID = "fits-recompute-status"
+
+
 @router.post("/augmented-parts/fits/recompute", response_class=HTMLResponse)
 async def recompute_all_fits(
     request: Request,
-    db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("fits:write")),
 ):
-    result = await fit_matching_service.recompute_all(db)
-    await audit_crud.log(
-        db,
-        actor=actor,
-        action="update",
-        resource_type="Fit",
-        resource_id=None,
-        description=(
-            f"Recomputed fits for all parts: {result.created} created, "
-            f"{result.removed} removed"
-        ),
-    )
-    return HTMLResponse(
-        f"Recompute complete: {result.created} created, {result.removed} removed."
+    actor_id = actor.id
+
+    async def _run(job: jobs.Job) -> None:
+        async with AsyncSessionLocal() as db:
+            result = await fit_matching_service.recompute_all(db)
+            actor_row = await user_crud.get(db, actor_id)
+            await audit_crud.log(
+                db,
+                actor=actor_row,
+                action="update",
+                resource_type="Fit",
+                resource_id=None,
+                description=(
+                    f"Recomputed fits for all parts: {result.created} created, "
+                    f"{result.removed} removed"
+                ),
+            )
+            job.status = "done"
+            job.message = (
+                f"Recompute complete: {result.created} created, "
+                f"{result.removed} removed."
+            )
+
+    job = jobs.start("fit_recompute_all", _run)
+    return templates.TemplateResponse(
+        request,
+        "admin/_job_progress_inline.html",
+        {
+            "container_id": _ALL_FITS_RECOMPUTE_STATUS_ID,
+            "poll_url": f"/admin/augmented-parts/fits/recompute/jobs/{job.id}",
+            "job": job,
+        },
     )
 
 
+@router.get(
+    "/augmented-parts/fits/recompute/jobs/{job_id}", response_class=HTMLResponse
+)
+async def recompute_all_fits_job_status(
+    request: Request,
+    job_id: str,
+    _: User = Depends(web_require("fits:write")),
+):
+    job = jobs.get(job_id)
+    if job is None:
+        return HTMLResponse("Recompute job not found — it may have expired.")
+    if job.status == "running":
+        return templates.TemplateResponse(
+            request,
+            "admin/_job_progress_inline.html",
+            {
+                "container_id": _ALL_FITS_RECOMPUTE_STATUS_ID,
+                "poll_url": f"/admin/augmented-parts/fits/recompute/jobs/{job_id}",
+                "job": job,
+            },
+        )
+    return HTMLResponse(job.message)
+
+
 # --- Augmented Part Notes -------------------------------------------------