diff --git a/.claude/settings.local.json b/.claude/settings.local.json index f3fbedd..cc71e59 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -30,7 +30,11 @@ "Bash(lsof -ti:8000 -sTCP:LISTEN)", "Bash(xargs -r kill)", "Bash(rm -f /tmp/fg_cookies.txt /private/tmp/claude-501/-Users-roddis-dev-armament-fitguide/7873fd9c-2f2b-4583-877e-72fc88718f24/scratchpad/smoke_sort.py)", - "Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://localhost:8000/login)" + "Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://localhost:8000/login)", + "Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)", + "Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")", + "Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")", + "Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")" ] } } diff --git a/app/services/jobs.py b/app/services/jobs.py new file mode 100644 index 0000000..271ceda --- /dev/null +++ b/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 diff --git a/app/templates/admin/_job_progress.html b/app/templates/admin/_job_progress.html new file mode 100644 index 0000000..64a5ff0 --- /dev/null +++ b/app/templates/admin/_job_progress.html @@ -0,0 +1,7 @@ +
+

⏳ {{ job.message }}

+
diff --git a/app/templates/admin/_job_progress_inline.html b/app/templates/admin/_job_progress_inline.html new file mode 100644 index 0000000..b46622d --- /dev/null +++ b/app/templates/admin/_job_progress_inline.html @@ -0,0 +1,5 @@ +⏳ {{ job.message }} diff --git a/app/templates/admin/_scope_import_result.html b/app/templates/admin/_scope_import_result.html index 4f7d858..bdd4fba 100644 --- a/app/templates/admin/_scope_import_result.html +++ b/app/templates/admin/_scope_import_result.html @@ -1,3 +1,4 @@ +
{% if error %}

{{ error }}

@@ -39,3 +40,4 @@ {% endif %}
{% endif %} +
diff --git a/app/web/admin.py b/app/web/admin.py index 29c4b5f..b30eb39 100644 --- a/app/web/admin.py +++ b/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 ------------------------------------------------- diff --git a/tests/test_admin_background_jobs.py b/tests/test_admin_background_jobs.py new file mode 100644 index 0000000..ba60476 --- /dev/null +++ b/tests/test_admin_background_jobs.py @@ -0,0 +1,156 @@ +import asyncio +import re + +import pytest + +from app.crud import role as role_crud +from app.models.attachment_point import AttachmentPoint +from app.models.attachment_point_name import AttachmentPointName +from app.models.augmented_part import AugmentedPart +from app.models.part_type import PartType +from tests.test_scope_import import _scope_row, _workbook + +_JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"') + + +async def _web_login(client, email, password="secret123"): + resp = await client.post("/login", data={"email": email, "password": password}) + assert resp.status_code == 303 + + +async def _poll_until_done(client, url, max_tries=50): + """Follow the same htmx polling loop a browser would, until no further + ``hx-get`` job-status link is present in the response.""" + body = "" + for _ in range(max_tries): + resp = await client.get(url) + assert resp.status_code == 200 + body = resp.text + match = _JOB_URL_RE.search(body) + if not match: + return body + url = match.group(1) + await asyncio.sleep(0.02) + raise AssertionError(f"job did not finish after {max_tries} polls: {body!r}") + + +@pytest.mark.asyncio +async def test_scope_import_runs_in_background( + client, seeded, make_user, db_sessionmaker, monkeypatch +): + monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + async with db_sessionmaker() as db: + perms = seeded["perms"] + editor = await role_crud.create( + db, + name="scope-import-editor", + permission_ids=[perms["augmented_parts:write"].id], + ) + editor_id = editor.id + await make_user("importer@example.com", role_ids=[editor_id]) + await _web_login(client, "importer@example.com") + + data = _workbook([_scope_row(product_number="BG-SCOPE-1")]) + resp = await client.post( + "/admin/scope-import", + files={"file": ("scopes.xlsx", data, "application/octet-stream")}, + ) + assert resp.status_code == 200 + assert "scope-import-status" in resp.text + match = _JOB_URL_RE.search(resp.text) + assert match, resp.text + + final_body = await _poll_until_done(client, match.group(1)) + assert "1 created" in final_body + + +@pytest.mark.asyncio +async def test_augmented_parts_sync_runs_in_background( + client, seeded, make_user, db_sessionmaker, monkeypatch +): + monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + async with db_sessionmaker() as db: + perms = seeded["perms"] + editor = await role_crud.create( + db, name="sync-editor", permission_ids=[perms["augmented_parts:write"].id] + ) + editor_id = editor.id + await make_user("syncer@example.com", role_ids=[editor_id]) + await _web_login(client, "syncer@example.com") + + resp = await client.post("/admin/augmented-parts/sync") + assert resp.status_code == 200 + assert 'id="augmented-part-list"' in resp.text + match = _JOB_URL_RE.search(resp.text) + assert match, resp.text + + final_body = await _poll_until_done(client, match.group(1)) + assert "Sync complete" in final_body + + +@pytest.mark.asyncio +async def test_fit_recompute_for_part_runs_in_background( + client, seeded, make_user, db_sessionmaker, monkeypatch +): + monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + async with db_sessionmaker() as db: + apn = AttachmentPointName(name="Objective") + part_t = PartType(name="Scope") + db.add_all([apn, part_t]) + await db.flush() + part = AugmentedPart( + description="Test scope", + colour_id=1, + manufacturer_id=1, + part_type_id=part_t.id, + ) + db.add(part) + await db.flush() + db.add( + AttachmentPoint(augmented_part_id=part.id, attachment_point_name_id=apn.id) + ) + await db.commit() + part_id = part.id + + perms = seeded["perms"] + editor = await role_crud.create( + db, + name="fit-recompute-editor", + permission_ids=[perms["fits:write"].id, perms["augmented_parts:read"].id], + ) + editor_id = editor.id + await make_user("recomputer@example.com", role_ids=[editor_id]) + await _web_login(client, "recomputer@example.com") + + resp = await client.post(f"/admin/augmented-parts/{part_id}/fits/recompute") + assert resp.status_code == 200 + assert f'id="ap-{part_id}-fits"' in resp.text + match = _JOB_URL_RE.search(resp.text) + assert match, resp.text + + final_body = await _poll_until_done(client, match.group(1)) + assert "created" in final_body and "removed" in final_body + + +@pytest.mark.asyncio +async def test_fit_recompute_all_runs_in_background( + client, seeded, make_user, db_sessionmaker, monkeypatch +): + monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker) + async with db_sessionmaker() as db: + perms = seeded["perms"] + editor = await role_crud.create( + db, name="fit-recompute-all-editor", permission_ids=[perms["fits:write"].id] + ) + editor_id = editor.id + await make_user("recomputerall@example.com", role_ids=[editor_id]) + await _web_login(client, "recomputerall@example.com") + + resp = await client.post("/admin/augmented-parts/fits/recompute") + assert resp.status_code == 200 + assert "fits-recompute-status" in resp.text + match = _JOB_URL_RE.search(resp.text) + assert match, resp.text + + final_body = await _poll_until_done(client, match.group(1)) + assert "Recompute complete" in final_body