diff --git a/Makefile b/Makefile index 27f448f..21e16e4 100644 --- a/Makefile +++ b/Makefile @@ -6,6 +6,10 @@ CELERY := $(VENV)/bin/celery ENV_FILE := src/config/env +# Suppress GNU Make's "Entering/Leaving directory" notices for recursive makes +# (e.g. `start` calling `run`); they name the same directory and only add noise. +MAKEFLAGS += --no-print-directory + .PHONY: setup-lite setup-full env env-lite update start run worker test help ## @@ -24,14 +28,14 @@ setup-lite: $(VENV)/.deps-installed env-lite @echo "[setup] Running database migrations..." @$(PYTHON) manage.py migrate @echo "" - @echo "Done. Run 'make run' to start the server." + @echo "Done. Run 'make start' to sync your library and start the server." ## setup-full — install full stack (PostgreSQL + Celery); run ./setup.sh first for OS deps setup-full: $(VENV)/.deps-installed env @echo "[setup] Running database migrations..." @$(PYTHON) manage.py migrate @echo "" - @echo "Done. Run 'make run' to start the server and 'make worker' to start the Celery worker." + @echo "Done. Start the Celery worker with 'make worker', then run 'make start' to sync your library and start the server." ## env — generate src/config/env from settings defaults (skips if already exists) env: @@ -93,7 +97,7 @@ update: @echo "[update] Running database migrations..." @$(PYTHON) manage.py migrate @echo "" - @echo "Done. Run 'make run' to start the server." + @echo "Done. Run 'make start' to sync your library and start the server." ## import PATH=… — import images from a directory (e.g. make import PATH=~/Pictures/Fujifilm) import: diff --git a/README.md b/README.md index fc72bb1..a8ee374 100644 --- a/README.md +++ b/README.md @@ -55,10 +55,11 @@ cd filmcase ```bash make setup-lite # creates venv, installs deps, generates SQLite config, runs migrations -make import PATH=/path/to/images # import your image collection make start # sync library and start the development server ``` +Then open the Library page and click **Add Folder** to import your photo collection. + --- ### Full install (for development and large collections) @@ -198,22 +199,16 @@ Python 3.11+ is required. --- -## Processing your image catalog - -Before using the web interface, you need to process your images so their EXIF data and recipe information are stored in the database. - -```bash -make import PATH=/path/to/your/images -``` - -The command behaves according to your install mode: +## Adding your images -- **Lite install** (`USE_ASYNC_TASKS=False`): images are processed one at a time in the foreground. The terminal blocks until all images are done. -- **Full install** (`USE_ASYNC_TASKS=True`): one Celery task is enqueued per image and processed in parallel by the worker. Start the worker first: +Register your photo folders in the **Library** and Filmcase imports them for you. Open +[http://localhost:8000/library/](http://localhost:8000/library/), click **Add Folder**, and +pick a directory. The images are imported straight away (in the background in lite mode, via +the Celery worker in full mode), and the folder is re-scanned on every `make start`, so new +photos are picked up automatically. - ```bash - make worker # or: celery -A src.config worker --loglevel=info --concurrency=8 - ``` +In full install mode, start the Celery worker first (`make worker`) so the import has +somewhere to run. --- @@ -238,9 +233,9 @@ If you only want to start the server without running a sync first, use `make run Visit `/images/` to see all processed images. Use the filter controls to narrow results by recipe, film simulation, white balance, and more. -### Process new images +### Add new images -Re-run `make import PATH=…` pointing at any directory containing new images. Already-processed images are updated in place with fresh EXIF data. Images without Fujifilm EXIF data are skipped. +Drop new files into a registered library folder and they are imported on the next `make start`. Adding a folder, or updating its path on the Library page, triggers an immediate sync of that folder. Already-known images are left as-is, and images without Fujifilm EXIF data are skipped. ### Rate images diff --git a/docs/ADRs/011-library-sync-on-folder-change.md b/docs/ADRs/011-library-sync-on-folder-change.md new file mode 100644 index 0000000..287ef5a --- /dev/null +++ b/docs/ADRs/011-library-sync-on-folder-change.md @@ -0,0 +1,214 @@ +# ADR 011 — Library sync on folder add/update + +**Status**: Accepted +**Date**: 2026-07-02 + +--- + +## Context + +ADR 010 introduced the Library: a list of `LibraryFolder` rows the app monitors, and a **startup** sync (`make start` runs `manage.py sync_library`) that walks every folder and imports new images. That covered the "detect new photos over time" need, but it deliberately did nothing while the app is running. + +The Library page (added alongside ADR 010) already lets the user add a folder, update a folder's path, and remove a folder, each through a use case. But those actions only change the monitored **list**. Nothing is imported until the next restart. A user who registers a folder full of photos sees it appear in the table and then... nothing happens, with no feedback and no obvious reason. The catalog silently lags behind the library until `make start` is run again. + +The app ships in two modes (ADR 003): **lite** (SQLite, no broker, `process_image` runs inline) and **full** (PostgreSQL + Celery, `process_image` runs in a worker). Any solution has to behave sensibly in both. + +--- + +## Problem + +Triggering the import from a web request is not as simple as calling the sync in the view — several forces make a naive implementation fail: + +- A folder's first import can be tens of thousands of files, so running it inside the request would hang the page or time out. +- The two install modes process images differently — inline in lite, enqueued to a worker in full — so a single trigger has to serve both. +- Once the work runs outside the request, its progress must live somewhere durable, so the UI can show it and it survives the user leaving the page. +- The image-processing code is shared with the manual `import` command, so sync-specific state must not leak into it. +- The process doing the work can die mid-run, leaving partial progress that must be recoverable. + +So: how do we run a potentially long import off the request, in both modes, with visible and recoverable progress, without contaminating the shared processing code? + +Removal is out of scope for importing: ADR 010 is add-only (the sync never deletes images), so removing a folder just deregisters it. + +--- + +## Options considered + +The central question is **how the triggered sync runs** relative to the HTTP request. + +### Option A — Run the sync synchronously inside the request + +The add/update view calls the sync directly and returns when it finishes. + +**Why we did not choose this option:** + +In full mode this is tolerable (the request only enqueues Celery tasks and returns), but in lite mode the request itself runs `process_image` for every new file. A first import of a large folder would hang the page for minutes and risk a proxy/browser timeout. It also fails problem (2): the work is tied to the request, so navigating away or a dropped connection could interrupt it. + +### Option C — Run the lite-mode sync in a detached subprocess + +Spawn `manage.py sync_library --folder ` as an independent OS process. + +**Why we did not choose this option:** + +It solves (1) and (2), and it even survives a web-process reload. But it is heavier: process spawning, argument plumbing, and status coordination that can only happen through the database anyway. The durability advantage over a thread — surviving a web reload — is already provided by the startup sync, which idempotently catches up on the next `make start`. The extra machinery is not justified for a single-user local app. + +### Option B — Decide the execution strategy in a use case; run lite in a background thread (chosen) + +A `trigger_folder_sync` **use case** decides how to run the sync based on the mode. In full mode it runs the single-folder sync inline (which only enqueues tasks and returns fast). In lite mode it hands the sync to a background daemon thread via a small `services/background.py` runner, so the request returns immediately while the thread processes images server-side. Progress is persisted in the database and polled over HTMX. + +**Why this was chosen:** + +- The thread lives in the web process, independent of the request, so it satisfies (2) directly: leaving the page only stops the polling, not the work. +- Putting the celery-vs-thread choice in a use case (not the view) keeps the decision in the application layer; the view just calls the use case, and raw threading stays behind a service, mirroring how `workertasks.enqueue_task` hides Celery. +- The startup sync remains the catch-up net, so the thread's one weakness — dying on a web-process reload — is already covered without a subprocess. + +--- + +## Decision + +- **Trigger only on add and path-update.** Remove stays pure deregistration. A path-update additionally resets `last_checked_at`, because the repointed tree is a different directory and mtime gating would otherwise skip its older subdirectories. + +- **Single-folder scope.** The per-folder walk-and-dispatch logic is factored into a `sync_folder` use case. `sync_library` (startup) becomes a loop over `sync_folder`, so an add never re-walks every registered folder. + +- **Execution strategy lives in `trigger_folder_sync`.** Full mode calls `sync_folder` directly (fast: it enqueues Celery tasks). Lite mode runs `sync_folder` in a daemon thread via `services/background.py`. The view only calls the use case. + +- **Progress lives in a `SyncRun` model, polled over HTMX.** Because the sync runs server-side, its state must be durable and readable from any request. Each run records `state` (`SCANNING` → `PROCESSING` → `COMPLETED`/`FAILED`/`INTERRUPTED`), `total`, `processed`, `skipped`, `errors`, and timestamps. History is kept; the latest run per folder is shown. A conditional `UniqueConstraint` allows **at most one active run per folder**, which doubles as the "already syncing" guard. + +- **Per-image work is composed, not coupled.** The domain `process_image` operation and the generic `process_image_task` (also used by the manual import command) stay sync-agnostic. A `process_synced_image` use case composes `process_image` with progress bookkeeping: it counts skips (`NoFilmSimulationError`) and errors, and finalises the run when every image is accounted for. Both interface adapters — a thin `sync_process_image_task` (full) and the thread loop (lite) — delegate to it. + +- **SQLite tuning for lite concurrency.** WAL and a busy timeout are enabled via `DATABASES["OPTIONS"]`, derived from `DB_ENGINE`. WAL lets the foreground request threads read while the background thread writes; the busy timeout makes a colliding foreground write wait rather than raise "database is locked". Per-image transactions plus JPEG-only fast hashing keep write-lock holds tiny, so rating/recipe writes are never starved. + +- **Crash recovery.** A run left `SCANNING`/`PROCESSING` when its process dies is marked `INTERRUPTED` at the start of the next `make start` sync, which then idempotently re-imports. + +- **Full-mode worker-down.** `trigger_folder_sync` pings for a worker up front; if none responds it surfaces a Library-page error and creates no run (no stuck badge). + +--- + +## Diagrams + +### Data model + +```mermaid +erDiagram + LibraryFolder { + int id PK + string path "normalized absolute, unique" + datetime last_processed_at + datetime last_checked_at "reset on path update" + } + SyncRun { + int id PK + int folder_id FK + string state "SCANNING | PROCESSING | COMPLETED | FAILED | INTERRUPTED" + int total "null while scanning" + int processed + int skipped + int errors + datetime started_at + datetime finished_at + } + Image { + int id PK + string filepath + } + + LibraryFolder ||--o{ SyncRun : "has runs (≤1 active)" + LibraryFolder ||..o{ Image : "monitors (no FK)" +``` + +### Full mode — add folder triggers an enqueue-and-return sync + +```mermaid +sequenceDiagram + actor User + participant View as LibraryFolderAdd view + participant Trigger as trigger_folder_sync uc + participant Sync as sync_folder uc + participant Worker as Celery worker(s) + participant PSI as process_synced_image uc + participant DB + + User->>View: POST /library/new/ + View->>Trigger: trigger_folder_sync(folder_id) + Trigger->>Trigger: worker reachable? (else error, no run) + Trigger->>Sync: sync_folder(folder_id) + Sync->>DB: start_sync_run (SCANNING) + Sync->>Sync: walk folder, diff vs known paths + Sync->>DB: begin_processing(total=N) + Sync-->>Worker: enqueue N sync_process_image tasks + Sync-->>View: return + View-->>User: redirect to /library/ + loop each task (concurrent) + Worker->>PSI: process_synced_image(path, run_id) + PSI->>DB: process_image + record_processed/skipped/error (F() atomic) + PSI->>DB: complete_sync_run if all accounted (conditional, one winner) + end + User->>View: folder row polls sync-status every 2s (HTMX) +``` + +### Lite mode — add folder triggers a background thread + +```mermaid +sequenceDiagram + actor User + participant View as LibraryFolderAdd view + participant Trigger as trigger_folder_sync uc + participant BG as background.run_in_background + participant Sync as sync_folder uc + participant PSI as process_synced_image uc + participant DB + + User->>View: POST /library/new/ + View->>Trigger: trigger_folder_sync(folder_id) + Trigger->>BG: run_in_background(sync_folder, folder_id) + Trigger-->>View: return + View-->>User: redirect to /library/ + Note over BG: daemon thread, outlives the request + BG->>Sync: sync_folder(folder_id) + Sync->>DB: start_sync_run (SCANNING) → begin_processing(total=N) + loop each new path (sequential) + Sync->>PSI: process_synced_image(path, run_id) + PSI->>DB: process_image + record progress + end + PSI->>DB: complete_sync_run (COMPLETED) + Note over User,DB: User navigates away and back. The thread keeps running
and the row re-reads DB state on the next poll. +``` + +--- + +## Progress tracking + +### Options considered + +**Option 1 — Introspect the Celery queue / result backend.** Derive progress in full mode from broker queue depth or `inspect()`, or from a `GroupResult.completed_count()`. + +*Rejected.* Broker/inspect counts are cluster-wide, not per-folder, and imperfect (queue depth vs reserved vs active); polling `inspect()` is a broadcast RPC. The clean `group`/`GroupResult` route needs a real result backend, but the app is configured with `rpc://`. None of it helps lite mode, which has no broker at all. + +**Option 2 — A dedicated `SyncRun` table (chosen).** Each task/thread reports against a per-folder run row. + +*Chosen.* It is per-folder-accurate, backend-agnostic, and identical across modes: lite's thread and full's tasks increment the **same** model, read by the **same** HTMX status endpoint. Under concurrent Celery workers, counters use atomic `F()` increments and completion is a conditional `UPDATE ... WHERE state = 'PROCESSING'` so exactly one finisher transitions the run and emits the completion event. + +--- + +## Consequences + +- New `SyncRun` model and migration; a `services/background.py` thread runner; `sync_folder`, `process_synced_image`, and `trigger_folder_sync` use cases; a `sync_process_image_task`; a sync-status view with HTMX polling in the folder row. +- `sync_library` becomes an all-folders loop over `sync_folder` that first interrupts dangling runs. The `manage.py sync_library` entry point and its result contract are unchanged. +- `process_image` and the generic `process_image_task` are untouched, so the manual `import` path is unaffected. +- Lite installs run with SQLite WAL enabled (a persistent, idempotent property of the database file). +- A new Celery task means the worker must be restarted after deploying this change before full-mode syncs will be processed; until then those messages are discarded and the next `make start` re-syncs. + +--- + +## Interface layer + +| Artifact | Location | +|---|---| +| Trigger use case | `src/application/usecases/library/trigger_folder_sync.py` | +| Single-folder sync use case | `src/application/usecases/library/sync_folder.py` | +| Per-image use case | `src/application/usecases/library/process_synced_image.py` | +| Background runner | `src/services/background.py` | +| Celery task | `sync_process_image_task` in `src/interfaces/tasks.py` | +| Status view | `LibraryFolderSyncStatus` → `library//sync-status/` | +| Status partial | `src/interfaces/templates/library/partials/sync_status.html` | + +The add and path-update views call `trigger_folder_sync` after the folder mutation and map `CeleryWorkerUnavailable` to a Library-page error. The folder row lazy-loads its status partial on load and, while a run is active, polls the status URL every 2 seconds, swapping to a terminal summary (which drops the poll trigger) when the run finishes. diff --git a/docs/index.md b/docs/index.md index 8fb1d5f..ce39280 100644 --- a/docs/index.md +++ b/docs/index.md @@ -10,6 +10,7 @@ - [Library Sync](library_sync.md) — how make start scans library folders, deduplicates against the catalog, and uses timestamps to skip unchanged directories - [EXIF Mapping](exif_mapping.md) — how Fujifilm EXIF fields map to database model fields - [Recipe Naming](recipe_naming.md) — how recipes are named and the constraints inherited from the camera +- [Recipe Graphs](recipe_graphs.md) — the film simulation graph and version-line graph views, and how to read node distance - [Image Matching](favorite_image_matching.md) — how images are matched to the catalogue when rating in bulk - [PTP Encodings](ptp_encodings.md) — PTP/USB encoding reference for camera communication @@ -31,4 +32,7 @@ - [ADR 006 — QR Decode Library and Minimum QR Code Size](ADRs/006-qr-decode-library-and-size.md) — QR decode library choice and minimum QR code size - [ADR 007 — Normalize Recipe Data Before Storage](ADRs/007-normalize-recipe-data.md) — normalizing recipe data before storage - [ADR 008 — Recipe Versioning via Generalised Grouping](ADRs/008-recipe-versioning.md) — version lines and recipe families via a shared grouping abstraction +- [ADR 009 — Moving a Recipe Between Version Lines](ADRs/009-move-recipe-between-version-lines.md) — reassigning an existing recipe to a different VERSION_LINE group while keeping positions contiguous +- [ADR 010 — Image Library: Folder Monitoring and Catalog Sync](ADRs/010-image-library.md) — persisting monitored folders and detecting and importing new images automatically at startup +- [ADR 011 — Library Sync on Folder Add/Update](ADRs/011-library-sync-on-folder-change.md) — triggering a single-folder sync from the Library page with server-side, DB-backed progress in both install modes diff --git a/docs/library_sync.md b/docs/library_sync.md index 1a3a4c1..bc2e330 100644 --- a/docs/library_sync.md +++ b/docs/library_sync.md @@ -43,6 +43,26 @@ missing path is reported in the command output and does not abort the sync. across the worker pool. If no Celery worker responds to a ping at the start of the sync, the entire sync is skipped with a warning. +## Syncing from the Library page + +You no longer have to restart the app to pick up a newly registered folder. Adding a folder, +or changing an existing folder's path, triggers a sync of that one folder straight away. +Removing a folder does not trigger anything: it only stops the folder being monitored and +never deletes images that were already imported. + +The triggered sync reuses the same per-folder scan described above and behaves according to +your install mode: + +- **Lite install:** the sync runs in a background thread, so the page responds immediately + while images are imported behind the scenes. You can navigate away and come back; the work + keeps running on the server. +- **Full install:** the new images are enqueued to the Celery worker and the page returns at + once. If no worker is reachable, the folder is still added but a message explains that it + could not be synced (start a worker with `make worker`, then re-add or re-save the folder). + +Changing a folder's path also clears its last-checked timestamp, so the whole new location is +rescanned from scratch. Progress appears live in the folder's **Sync** column (see below). + ## Timestamp-based directory gating For large collections, walking every subdirectory on every startup would be slow. To avoid @@ -68,3 +88,9 @@ Each folder row in the Library page shows two timestamps: whether anything new was found. - **Last Synced** -- the most recent time the sync actually imported or enqueued new images from this folder. This stays blank until at least one new file is found. + +The **Sync** column shows the status of the most recent sync for each folder: `Scanning...` +while the folder is being walked, a progress bar while images are imported, and a final +summary such as `Imported 36, skipped 3` when it finishes. While a sync is active, the column +refreshes on its own every couple of seconds, so you can watch it progress without reloading +the page. diff --git a/src/application/usecases/library/dataclasses.py b/src/application/usecases/library/dataclasses.py index 4e1c8ca..afc8371 100644 --- a/src/application/usecases/library/dataclasses.py +++ b/src/application/usecases/library/dataclasses.py @@ -12,6 +12,22 @@ class LibraryFolderData: last_checked_at: datetime | None +@attrs.frozen +class SyncRunData: + folder_id: int + total: int | None + processed: int + skipped: int + errors: int + percent: int + is_active: bool + is_scanning: bool + is_processing: bool + is_completed: bool + is_failed: bool + is_interrupted: bool + + @attrs.frozen class FilesystemEntry: name: str diff --git a/src/application/usecases/library/process_synced_image.py b/src/application/usecases/library/process_synced_image.py new file mode 100644 index 0000000..0531e8a --- /dev/null +++ b/src/application/usecases/library/process_synced_image.py @@ -0,0 +1,41 @@ +import structlog + +from src.domain.images import operations as image_operations +from src.domain.images.queries import NoFilmSimulationError +from src.domain.library import operations as library_operations +from src.domain.library import queries as library_queries + +logger = structlog.get_logger("application.library.process_synced_image") + + +def process_synced_image(*, image_path: str, sync_run_id: int) -> None: + """ + Process a single image as part of sync run *sync_run_id* and record progress. + + Composes the pure image-processing operation with sync-run bookkeeping so the + processing operation stays unaware of sync: non-Fujifilm files count as + skipped, unexpected failures are logged and counted as errors (the run + continues), and successful imports count as processed. When every image in the + run has a terminal outcome, the run is completed. + + If the run no longer exists (its folder was removed while this work was + queued), the call returns without processing. + """ + try: + run = library_queries.get_sync_run(run_id=sync_run_id) + except library_queries.SyncRunNotFound: + return + + try: + image_operations.process_image(image_path=image_path) + except NoFilmSimulationError: + run.record_skipped() + except Exception: + logger.exception("Failed to process image during sync") + run.record_error() + else: + run.record_processed() + + run.refresh_from_db() + if run.all_images_accounted_for(): + library_operations.complete_sync_run(run=run) diff --git a/src/application/usecases/library/sync_folder.py b/src/application/usecases/library/sync_folder.py new file mode 100644 index 0000000..2a48277 --- /dev/null +++ b/src/application/usecases/library/sync_folder.py @@ -0,0 +1,71 @@ +from datetime import datetime, timezone + +from django.conf import settings + +from src.application.usecases.library.process_synced_image import process_synced_image +from src.domain.images import queries as image_queries +from src.domain.library import operations as library_operations +from src.domain.library import queries as library_queries +from src.services import workertasks + +_SYNC_PROCESS_IMAGE_TASK = "src.interfaces.tasks.sync_process_image_task" + + +def sync_folder(*, folder_id: int) -> None: + """ + Scan a single library folder and import new images, tracking progress in a + SyncRun. + + Creates a run, walks the folder (mtime-gated by its last_checked_at), and + dispatches each new image: in async mode by enqueuing a Celery task, in sync + mode by processing inline. The run is completed here only when nothing new is + found; otherwise the last processed image completes it. + + Returns without doing anything if the folder no longer exists or already has + an active run (the concurrency guard). + """ + try: + folder = library_queries.get_library_folder(folder_id=folder_id) + except library_queries.LibraryFolderNotFound: + return + + try: + run = library_operations.start_sync_run(folder=folder) + except library_operations.SyncAlreadyInProgress: + return + + # last_checked_at records when the scan started, so files added during or + # after this scan are still caught on the next run. + now = datetime.now(tz=timezone.utc) + + try: + found_paths = image_queries.get_image_paths_in_folder( + folder_path=folder.path, + last_checked_at=folder.last_checked_at, + ) + except FileNotFoundError: + folder.set_last_checked_at(value=now) + library_operations.fail_sync_run(run=run, message="Folder does not exist") + return + + known_paths = image_queries.get_all_known_image_paths() + new_paths = sorted(set(found_paths) - known_paths) + + run.begin_processing(total=len(new_paths)) + folder.set_last_checked_at(value=now) + if new_paths: + folder.set_last_processed_at(value=now) + + if not new_paths: + library_operations.complete_sync_run(run=run) + return + + for path in new_paths: + if settings.USE_ASYNC_TASKS: + workertasks.enqueue_task( + task_name=_SYNC_PROCESS_IMAGE_TASK, + kwargs={"image_path": path, "sync_run_id": run.pk}, + queue=settings.PROCESS_IMAGE_QUEUE, + ) + else: + process_synced_image(image_path=path, sync_run_id=run.pk) diff --git a/src/application/usecases/library/sync_library.py b/src/application/usecases/library/sync_library.py index e131ecc..956a29d 100644 --- a/src/application/usecases/library/sync_library.py +++ b/src/application/usecases/library/sync_library.py @@ -1,11 +1,9 @@ import attrs -from datetime import datetime, timezone from django.conf import settings -from src.domain.images import operations as image_operations -from src.domain.images import queries as image_queries -from src.domain.images.queries import NoFilmSimulationError +from src.application.usecases.library.sync_folder import sync_folder +from src.domain.library import operations as library_operations from src.domain.library import queries as library_queries from src.services import workertasks @@ -27,13 +25,16 @@ class SyncLibraryResult: def sync_library() -> SyncLibraryResult: """ - Scan all registered library folders and import new images into the catalog. + Scan every registered library folder and import new images into the catalog. - Loads all known image paths in a single DB query, then walks each folder - and processes only paths not yet in the catalog. Paths that appear in - multiple overlapping folders are deduplicated across the entire sync run. + Recovers any runs abandoned by a previous process (marking them interrupted), + then syncs each folder in turn via the single-folder use case. In async mode, + checks for a reachable Celery worker before doing any work. - In async mode, checks for a reachable Celery worker before doing any work. + The result aggregates per-folder outcomes from each folder's sync run: in async + mode ``new_files_found`` counts images enqueued for processing, in sync mode it + counts images actually imported. Folders that no longer exist on disk are + reported in ``missing_folders``. :raises CeleryWorkerUnavailable: If USE_ASYNC_TASKS is True and no Celery worker responds within the ping timeout. @@ -41,49 +42,25 @@ def sync_library() -> SyncLibraryResult: if settings.USE_ASYNC_TASKS and not workertasks.is_celery_worker_available(): raise CeleryWorkerUnavailable() - known_paths = image_queries.get_all_known_image_paths() - folders = library_queries.get_all_library_folders() + library_operations.interrupt_active_sync_runs() - all_found_paths: set[str] = set() + folders = library_queries.get_all_library_folders() new_files_found = 0 skipped_non_fujifilm = 0 missing_folders: list[str] = [] - now = datetime.now(tz=timezone.utc) for folder in folders: - try: - found_paths = image_queries.get_image_paths_in_folder( - folder_path=folder.path, - last_checked_at=folder.last_checked_at, - ) - except FileNotFoundError: - missing_folders.append(folder.path) - folder.set_last_checked_at(value=now) + sync_folder(folder_id=folder.pk) + run = library_queries.get_latest_sync_run(folder_id=folder.pk) + if run is None: continue - - new_in_folder = set(found_paths) - known_paths - all_found_paths - all_found_paths |= set(found_paths) - - processed_in_folder = 0 - for path in new_in_folder: - if settings.USE_ASYNC_TASKS: - workertasks.enqueue_task( - task_name="src.interfaces.tasks.process_image_task", - kwargs={"image_path": path}, - queue=settings.PROCESS_IMAGE_QUEUE, - ) - processed_in_folder += 1 - else: - try: - image_operations.process_image(image_path=path) - processed_in_folder += 1 - except NoFilmSimulationError: - skipped_non_fujifilm += 1 - - new_files_found += processed_in_folder - folder.set_last_checked_at(value=now) - if processed_in_folder > 0: - folder.set_last_processed_at(value=now) + if run.state == run.STATE_FAILED: + missing_folders.append(folder.path) + elif settings.USE_ASYNC_TASKS: + new_files_found += run.total or 0 + else: + new_files_found += run.processed + skipped_non_fujifilm += run.skipped return SyncLibraryResult( folders_scanned=len(folders), diff --git a/src/application/usecases/library/trigger_folder_sync.py b/src/application/usecases/library/trigger_folder_sync.py new file mode 100644 index 0000000..4cfa47d --- /dev/null +++ b/src/application/usecases/library/trigger_folder_sync.py @@ -0,0 +1,25 @@ +from django.conf import settings + +from src.application.usecases.library.sync_folder import sync_folder +from src.application.usecases.library.sync_library import CeleryWorkerUnavailable as CeleryWorkerUnavailable +from src.services import background, workertasks + + +def trigger_folder_sync(*, folder_id: int) -> None: + """ + Trigger a sync of a single folder from a web request. + + Decides how to run the sync so the request returns promptly. In async mode it + verifies a Celery worker is reachable and then runs the sync inline (which only + enqueues per-image tasks, so it is fast). In sync mode it runs the sync in a + background thread, so image processing does not block the request. + + :raises CeleryWorkerUnavailable: If USE_ASYNC_TASKS is True and no Celery + worker responds; no run is created in that case. + """ + if settings.USE_ASYNC_TASKS: + if not workertasks.is_celery_worker_available(): + raise CeleryWorkerUnavailable() + sync_folder(folder_id=folder_id) + else: + background.run_in_background(sync_folder, folder_id=folder_id) diff --git a/src/config/settings.py b/src/config/settings.py index d502524..f118e1e 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -31,7 +31,7 @@ DB_HOST: str = env.str("DB_HOST", default="127.0.0.1") DB_PORT: str = env.str("DB_PORT", default="5432") -DATABASES = { +DATABASES: dict[str, dict[str, object]] = { "default": { "ENGINE": DB_ENGINE, "NAME": DB_NAME, @@ -42,6 +42,16 @@ } } +if DB_ENGINE.endswith("sqlite3"): + # Lite mode runs SQLite with a background sync thread writing while the web + # request threads read. WAL lets readers proceed alongside the single writer; + # the busy timeout makes a colliding writer wait rather than raise "database + # is locked". journal_mode is persistent on the file (idempotent to re-set). + DATABASES["default"]["OPTIONS"] = { + "timeout": 5, # SQLite busy_timeout, applied per connection + "init_command": "PRAGMA journal_mode=WAL;", + } + DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" PTP_DEVICE: str = env.str("PTP_DEVICE", default="src.domain.camera.ptp_usb_device.PTPUSBDevice") # dotted import path to the PTP device implementation; swap for a stub/mock in tests diff --git a/src/data/migrations/0034_syncrun.py b/src/data/migrations/0034_syncrun.py new file mode 100644 index 0000000..d1cc149 --- /dev/null +++ b/src/data/migrations/0034_syncrun.py @@ -0,0 +1,36 @@ +# Generated by Django 6.0.3 on 2026-07-02 15:45 + +import django.db.models.deletion +import django.utils.timezone +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('data', '0033_libraryfolder'), + ] + + operations = [ + migrations.CreateModel( + name='SyncRun', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('state', models.CharField(max_length=16)), + ('total', models.IntegerField(null=True)), + ('processed', models.IntegerField(default=0)), + ('skipped', models.IntegerField(default=0)), + ('errors', models.IntegerField(default=0)), + ('error_message', models.TextField(null=True)), + ('started_at', models.DateTimeField(default=django.utils.timezone.now)), + ('finished_at', models.DateTimeField(null=True)), + ('created_at', models.DateTimeField(default=django.utils.timezone.now)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('folder', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='sync_runs', to='data.libraryfolder')), + ], + options={ + 'indexes': [models.Index(fields=['folder', '-started_at'], name='idx_sync_run_folder_started')], + 'constraints': [models.UniqueConstraint(condition=models.Q(('state__in', ('SCANNING', 'PROCESSING'))), fields=('folder',), name='unique_active_sync_run_per_folder')], + }, + ), + ] diff --git a/src/data/models/__init__.py b/src/data/models/__init__.py index be09f0d..1a240b6 100644 --- a/src/data/models/__init__.py +++ b/src/data/models/__init__.py @@ -8,6 +8,7 @@ RecipeGroupMember, Sensor, ) +from ._sync_run import SyncRun __all__ = [ "RECIPE_FIELDS", @@ -20,4 +21,5 @@ "RecipeGroup", "RecipeGroupMember", "Sensor", + "SyncRun", ] diff --git a/src/data/models/_library.py b/src/data/models/_library.py index 9004285..64ce591 100644 --- a/src/data/models/_library.py +++ b/src/data/models/_library.py @@ -41,5 +41,9 @@ def set_last_checked_at(self, *, value: datetime) -> None: self.last_checked_at = value self.save(update_fields=["last_checked_at", "updated_at"]) + def clear_last_checked_at(self) -> None: + self.last_checked_at = None + self.save(update_fields=["last_checked_at", "updated_at"]) + def __str__(self) -> str: return f"#{self.id} {self.path}" diff --git a/src/data/models/_sync_run.py b/src/data/models/_sync_run.py new file mode 100644 index 0000000..ad18021 --- /dev/null +++ b/src/data/models/_sync_run.py @@ -0,0 +1,114 @@ +from django.db import models +from django.utils import timezone + +from ._library import LibraryFolder + +_STATE_SCANNING = "SCANNING" +_STATE_PROCESSING = "PROCESSING" +_STATE_COMPLETED = "COMPLETED" +_STATE_FAILED = "FAILED" +_STATE_INTERRUPTED = "INTERRUPTED" + +# A run is "active" while it is scanning or processing; at most one active run is +# allowed per folder (enforced by a conditional UniqueConstraint below). +_ACTIVE_STATES = (_STATE_SCANNING, _STATE_PROCESSING) + +_STATE_MAX_LEN = 16 + + +class SyncRun(models.Model): + STATE_SCANNING = _STATE_SCANNING + STATE_PROCESSING = _STATE_PROCESSING + STATE_COMPLETED = _STATE_COMPLETED + STATE_FAILED = _STATE_FAILED + STATE_INTERRUPTED = _STATE_INTERRUPTED + ACTIVE_STATES = _ACTIVE_STATES + + folder = models.ForeignKey( + LibraryFolder, + on_delete=models.CASCADE, + related_name="sync_runs", + ) + state = models.CharField(max_length=_STATE_MAX_LEN) + total = models.IntegerField(null=True) # unknown during the scanning phase + processed = models.IntegerField(default=0) + skipped = models.IntegerField(default=0) + errors = models.IntegerField(default=0) + error_message = models.TextField(null=True) + started_at = models.DateTimeField(default=timezone.now) + finished_at = models.DateTimeField(null=True) + created_at = models.DateTimeField(default=timezone.now) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + indexes = [ + models.Index(fields=["folder", "-started_at"], name="idx_sync_run_folder_started"), + ] + constraints = [ + models.UniqueConstraint( + fields=["folder"], + condition=models.Q(state__in=_ACTIVE_STATES), + name="unique_active_sync_run_per_folder", + ), + ] + + # Factories + + @classmethod + def create(cls, *, folder: LibraryFolder) -> "SyncRun": + return cls.objects.create(folder=folder, state=_STATE_SCANNING) + + # Mutators + + def begin_processing(self, *, total: int) -> None: + self.state = _STATE_PROCESSING + self.total = total + self.save(update_fields=["state", "total", "updated_at"]) + + def record_processed(self) -> None: + self._increment("processed") + + def record_skipped(self) -> None: + self._increment("skipped") + + def record_error(self) -> None: + self._increment("errors") + + def _increment(self, field: str) -> None: + # Atomic increment so concurrent Celery tasks reporting against the same + # run never lose a count. The caller must refresh_from_db() to read the + # updated value. + type(self).objects.filter(pk=self.pk).update( + updated_at=timezone.now(), + **{field: models.F(field) + 1}, + ) + + def mark_completed(self) -> bool: + # Conditional so exactly one caller wins the finalize under concurrency. + # Returns True if this call transitioned the run to COMPLETED. + now = timezone.now() + rows = type(self).objects.filter(pk=self.pk, state=_STATE_PROCESSING).update( + state=_STATE_COMPLETED, + finished_at=now, + updated_at=now, + ) + return rows > 0 + + def mark_failed(self, *, message: str) -> None: + self.state = _STATE_FAILED + self.error_message = message + self.finished_at = timezone.now() + self.save(update_fields=["state", "error_message", "finished_at", "updated_at"]) + + # Queries + + def all_images_accounted_for(self) -> bool: + """ + Return True once every image in the run has a terminal outcome. + """ + if self.total is None: + return False + return self.processed + self.skipped + self.errors >= self.total + + def __str__(self) -> str: + return f"#{self.id} {self.state} folder #{self.folder_id}" diff --git a/src/domain/library/events.py b/src/domain/library/events.py index 5dac612..7888150 100644 --- a/src/domain/library/events.py +++ b/src/domain/library/events.py @@ -7,6 +7,10 @@ LIBRARY_FOLDER_ADDED = "library.folder.added" LIBRARY_FOLDER_REMOVED = "library.folder.removed" LIBRARY_FOLDER_PATH_UPDATED = "library.folder.path.updated" +LIBRARY_SYNC_RUN_STARTED = "library.sync.run.started" +LIBRARY_SYNC_RUN_COMPLETED = "library.sync.run.completed" +LIBRARY_SYNC_RUN_FAILED = "library.sync.run.failed" +LIBRARY_SYNC_RUN_INTERRUPTED = "library.sync.run.interrupted" def publish_event(*, event_type: str, **kwargs: object) -> None: diff --git a/src/domain/library/operations.py b/src/domain/library/operations.py index 75252e6..80dd424 100644 --- a/src/domain/library/operations.py +++ b/src/domain/library/operations.py @@ -2,6 +2,7 @@ from pathlib import Path from django.db import IntegrityError, transaction +from django.utils import timezone from src.data import models from src.domain.library import events @@ -17,6 +18,15 @@ class FolderAlreadyInLibrary(Exception): path: str +@attrs.frozen +class SyncAlreadyInProgress(Exception): + """ + Raised when a sync run is started for a folder that already has an active run. + """ + + folder_id: int + + def _normalize_path(path: str) -> str: return str(Path(path).expanduser().resolve()) @@ -91,9 +101,89 @@ def update_library_folder_path(*, folder_id: int, path: str) -> models.LibraryFo except IntegrityError: raise FolderAlreadyInLibrary(path=normalized) + # The new path is a different tree, so the previous scan timestamp no longer + # applies. Clearing it forces a full rescan (mtime gating would otherwise skip + # directories older than the old last_checked_at). + folder.clear_last_checked_at() + events.publish_event( event_type=events.LIBRARY_FOLDER_PATH_UPDATED, folder_id=folder.pk, path=folder.path, ) return folder + + +def start_sync_run(*, folder: models.LibraryFolder) -> models.SyncRun: + """ + Create a new sync run for *folder* in the scanning state. + + :raises SyncAlreadyInProgress: If *folder* already has an active (scanning or + processing) run. + """ + try: + with transaction.atomic(): + run = models.SyncRun.create(folder=folder) + except IntegrityError: + raise SyncAlreadyInProgress(folder_id=folder.pk) + + events.publish_event( + event_type=events.LIBRARY_SYNC_RUN_STARTED, + run_id=run.pk, + folder_id=folder.pk, + ) + return run + + +def complete_sync_run(*, run: models.SyncRun) -> bool: + """ + Mark *run* as completed if it is still processing. + + Uses a conditional update so that, under concurrent workers, exactly one + caller transitions the run and publishes the completion event. Returns True + if this call completed the run. + """ + completed = run.mark_completed() + if completed: + events.publish_event( + event_type=events.LIBRARY_SYNC_RUN_COMPLETED, + run_id=run.pk, + folder_id=run.folder_id, + ) + return completed + + +def fail_sync_run(*, run: models.SyncRun, message: str) -> None: + """ + Mark *run* as failed, recording *message* as the failure reason. + """ + run.mark_failed(message=message) + events.publish_event( + event_type=events.LIBRARY_SYNC_RUN_FAILED, + run_id=run.pk, + folder_id=run.folder_id, + reason=message, + ) + + +def interrupt_active_sync_runs() -> int: + """ + Mark every active (scanning or processing) sync run as interrupted. + + Called at startup to recover runs abandoned by a killed process, so no run + is left permanently active. Returns the number of runs interrupted. + """ + now = timezone.now() + count = models.SyncRun.objects.filter( + state__in=models.SyncRun.ACTIVE_STATES, + ).update( + state=models.SyncRun.STATE_INTERRUPTED, + finished_at=now, + updated_at=now, + ) + if count: + events.publish_event( + event_type=events.LIBRARY_SYNC_RUN_INTERRUPTED, + count=count, + ) + return count diff --git a/src/domain/library/queries.py b/src/domain/library/queries.py index 9283c72..7d81f84 100644 --- a/src/domain/library/queries.py +++ b/src/domain/library/queries.py @@ -23,6 +23,15 @@ class FolderNotFound(Exception): path: str +@attrs.frozen +class SyncRunNotFound(Exception): + """ + Raised when no SyncRun row matches the given run_id. + """ + + run_id: int + + def get_all_library_folders() -> list[models.LibraryFolder]: """ Return all registered library folders ordered by path. @@ -59,3 +68,44 @@ def list_subdirectories(*, path: str) -> tuple[str, ...]: if entry.is_dir() and not entry.name.startswith(".") ) return tuple(entries) + + +def get_latest_sync_run(*, folder_id: int) -> models.SyncRun | None: + """ + Return the most recently started sync run for *folder_id*, or None if the + folder has never been synced. + """ + return ( + models.SyncRun.objects.filter(folder_id=folder_id) + .order_by("-started_at", "-id") + .first() + ) + + +def get_sync_run(*, run_id: int) -> models.SyncRun: + """ + Return the SyncRun with the given id. + + :raises SyncRunNotFound: If no run with *run_id* exists (e.g. the folder was + removed while a task for this run was still queued). + """ + try: + return models.SyncRun.objects.get(pk=run_id) + except models.SyncRun.DoesNotExist: + raise SyncRunNotFound(run_id=run_id) + + +def get_active_sync_run(*, folder_id: int) -> models.SyncRun | None: + """ + Return the in-progress (scanning or processing) sync run for *folder_id*, or + None if no run is currently active. At most one active run can exist per + folder (enforced by a database constraint). + """ + return ( + models.SyncRun.objects.filter( + folder_id=folder_id, + state__in=models.SyncRun.ACTIVE_STATES, + ) + .order_by("-started_at", "-id") + .first() + ) diff --git a/src/interfaces/library/urls.py b/src/interfaces/library/urls.py index 5767d3e..dbd4ee7 100644 --- a/src/interfaces/library/urls.py +++ b/src/interfaces/library/urls.py @@ -8,4 +8,5 @@ path("library/browse/partial/", views.FilesystemBrowser.as_view(), name="library-browse"), path("library//delete/", views.LibraryFolderRemove.as_view(), name="library-folder-delete"), path("library//edit/", views.LibraryFolderPathUpdate.as_view(), name="library-folder-edit"), + path("library//sync-status/", views.LibraryFolderSyncStatus.as_view(), name="library-folder-sync-status"), ] diff --git a/src/interfaces/library/views.py b/src/interfaces/library/views.py index f873ce9..826ffae 100644 --- a/src/interfaces/library/views.py +++ b/src/interfaces/library/views.py @@ -5,6 +5,7 @@ from src.application.usecases.library import browse_filesystem as browse_filesystem_uc from src.application.usecases.library import dataclasses as library_dataclasses from src.application.usecases.library import remove_library_folder as remove_library_folder_uc +from src.application.usecases.library import trigger_folder_sync as trigger_folder_sync_uc from src.application.usecases.library import update_library_folder_path as update_library_folder_path_uc from src.data import models from src.domain.library import queries as domain_queries @@ -24,6 +25,26 @@ def _list_all_folders() -> list[library_dataclasses.LibraryFolderData]: return [_folder_data(f) for f in domain_queries.get_all_library_folders()] +def _sync_status(run: models.SyncRun) -> library_dataclasses.SyncRunData: + total = run.total + handled = run.processed + run.skipped + run.errors + percent = int(handled / total * 100) if total else 0 + return library_dataclasses.SyncRunData( + folder_id=run.folder_id, + total=total, + processed=run.processed, + skipped=run.skipped, + errors=run.errors, + percent=percent, + is_active=run.state in models.SyncRun.ACTIVE_STATES, + is_scanning=run.state == models.SyncRun.STATE_SCANNING, + is_processing=run.state == models.SyncRun.STATE_PROCESSING, + is_completed=run.state == models.SyncRun.STATE_COMPLETED, + is_failed=run.state == models.SyncRun.STATE_FAILED, + is_interrupted=run.state == models.SyncRun.STATE_INTERRUPTED, + ) + + class LibraryFolderList(generic.View): """Display the list of monitored library folders.""" @@ -39,7 +60,7 @@ def post(self, request: http.HttpRequest) -> http.HttpResponse: if not path: return http.HttpResponseBadRequest("path is required") try: - add_library_folder_uc.add_library_folder(path=path) + folder = add_library_folder_uc.add_library_folder(path=path) except add_library_folder_uc.FolderNotFound as exc: return shortcuts.render(request, "library/library.html", { "folders": _list_all_folders(), @@ -50,6 +71,14 @@ def post(self, request: http.HttpRequest) -> http.HttpResponse: "folders": _list_all_folders(), "error": f"Folder is already in the library: {exc.path}", }) + + try: + trigger_folder_sync_uc.trigger_folder_sync(folder_id=folder.folder_id) + except trigger_folder_sync_uc.CeleryWorkerUnavailable: + return shortcuts.render(request, "library/library.html", { + "folders": _list_all_folders(), + "error": "Folder added, but no image worker is running to sync it. Start one with 'make worker'.", + }) return shortcuts.redirect(urls.reverse("library-list")) @@ -91,9 +120,29 @@ def post(self, request: http.HttpRequest, folder_id: int) -> http.HttpResponse: "folders": _list_all_folders(), "error": f"Folder is already in the library: {exc.path}", }) + + try: + trigger_folder_sync_uc.trigger_folder_sync(folder_id=folder_id) + except trigger_folder_sync_uc.CeleryWorkerUnavailable: + return shortcuts.render(request, "library/library.html", { + "folders": _list_all_folders(), + "error": "Path updated, but no image worker is running to sync it. Start one with 'make worker'.", + }) return shortcuts.redirect(urls.reverse("library-list")) +class LibraryFolderSyncStatus(generic.View): + """Return an HTMX partial with the latest sync-run status for a folder.""" + + def get(self, request: http.HttpRequest, folder_id: int) -> http.HttpResponse: + run = domain_queries.get_latest_sync_run(folder_id=folder_id) + status = _sync_status(run) if run is not None else None + return shortcuts.render(request, "library/partials/sync_status.html", { + "status": status, + "folder_id": folder_id, + }) + + class FilesystemBrowser(generic.View): """Return an HTMX partial for the filesystem browser. diff --git a/src/interfaces/tasks.py b/src/interfaces/tasks.py index 516ccac..b85db14 100644 --- a/src/interfaces/tasks.py +++ b/src/interfaces/tasks.py @@ -4,6 +4,7 @@ from celery import shared_task from django.conf import settings +from src.application.usecases.library.process_synced_image import process_synced_image from src.domain.images import events, operations from src.domain.images.thumbnails import operations as thumbnail_operations @@ -31,6 +32,16 @@ def process_image_task(self: Any, /, *, image_path: str, **kwargs: object) -> st return f"Processed {recipe.filename}" +@shared_task(name="library.sync_process_image", bind=True, queue=settings.PROCESS_IMAGE_QUEUE) +def sync_process_image_task(self: Any, /, *, image_path: str, sync_run_id: int, **kwargs: object) -> str: + """ + Celery task that processes a single image for a library sync run and reports + progress against the run. + """ + process_synced_image(image_path=image_path, sync_run_id=sync_run_id) + return f"Processed {image_path} for sync run {sync_run_id}" + + @shared_task(name="domain.generate_thumbnail", bind=True, queue=settings.PROCESS_IMAGE_QUEUE) def generate_thumbnail_task(self: Any, /, *, filepath: str, width: int, **kwargs: object) -> str: """ diff --git a/src/interfaces/templates/library/includes/folder_row.html b/src/interfaces/templates/library/includes/folder_row.html index c9c3005..d71417a 100644 --- a/src/interfaces/templates/library/includes/folder_row.html +++ b/src/interfaces/templates/library/includes/folder_row.html @@ -14,6 +14,13 @@ Never {% endif %} + +
+