diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 8f777144..b615f3dd 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -172,6 +172,7 @@ export default defineConfig({ { text: "User lifecycle", link: "/architecture/user-lifecycle" }, { text: "Authentication model", link: "/architecture/auth-model" }, { text: "Magic-link auth", link: "/architecture/magic-link-auth" }, + { text: "Background jobs", link: "/architecture/jobs" }, ], }, { text: "FAQ", link: "/faq" }, diff --git a/docs/architecture/index.md b/docs/architecture/index.md index ae7aa303..29df37d9 100644 --- a/docs/architecture/index.md +++ b/docs/architecture/index.md @@ -74,3 +74,4 @@ src/ - [Caching Architecture →](/architecture/caching) - [Resource Listing API →](/architecture/resource-listing) - [Storage Quotas →](/architecture/storage-quotas) +- [Background Jobs →](/architecture/jobs) diff --git a/docs/architecture/jobs.md b/docs/architecture/jobs.md new file mode 100644 index 00000000..5195237e --- /dev/null +++ b/docs/architecture/jobs.md @@ -0,0 +1,246 @@ +# Background Jobs + +OxiCloud runs periodic maintenance work through a central **JobRegistry** +scheduler. This document is for implementors adding a new tenant or +debugging an existing one. + +The design rationale (why a registry vs. per-service `tokio::spawn` +loops, the two-engine split, plugin future) lives in the plan doc: +[`docs/plan/job-registry.md`](../plan/job-registry.md). This page +is the "how to plug in" reference. + +--- + +## When does a background loop belong here? + +Single decision question: + +> **"Would an operator plausibly `POST /api/admin/jobs/{name}/trigger` +> to make it run right now?"** + +If yes, register it with JobRegistry. You get: + +- Uniform admin surface (list, trigger, last-outcome). +- Uniform `oxicloud::scheduler` log line per run with `elapsed_ms`. +- Panic containment (a handler that panics doesn't kill the scheduler). +- Exclusivity: only one in-flight run per job name (a tick that fires + while the previous run is still going is skipped, with a warning). +- Optional wall-clock timeout. +- Optional acceleration parameter (`?force=true` → `JobRunArgs.force`). + +If no — the loop is a queue drainer (`tree_etag_flush_job`), a +continuous event reactor (`content_index_worker`), or a stats printer +(`db_pool_monitor`) — keep it as its own dedicated `tokio::spawn` +loop. Wedging it into JobRegistry adds framework overhead for no +operator benefit. + +--- + +## Current tenants + +| Job name | Cadence | Force semantic | Service | +|---|---|---|---| +| `trash_cleanup` | 24 h (hardcoded in DI, no env var yet) | ignored | [`trash_cleanup_service.rs`](../../src/infrastructure/services/trash_cleanup_service.rs) | +| `storage_reconcile`| `OXICLOUD_STORAGE_USAGE_RECONCILE_SECS` (default 600s, min 30s) | ignored | [`storage_usage_service.rs`](../../src/application/services/storage_usage_service.rs) | +| `dedup_gc` | on-demand only (trash cleanup runs it inline as its tail step) | `force=true` → `garbage_collect_force()` (skip orphan grace) | [`dedup_service.rs`](../../src/infrastructure/services/dedup_service.rs) | +| `grant_cleanup` | `OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS` (default 24h) — feature-gated by `OXICLOUD_GRANT_CLEANUP_ENABLED` | `force=true` → `purge(Some(0))` (grace_days=0) | [`grant_cleanup_service.rs`](../../src/infrastructure/services/grant_cleanup_service.rs) | + +--- + +## Adding a new job — recipe + +### 1. Implement `JobHandler` on your service + +```rust +use std::sync::Arc; +use async_trait::async_trait; + +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs}; + +pub const MY_JOB_NAME: &str = "my_job"; // stable snake_case + +#[async_trait] +impl JobHandler for MyService { + fn name(&self) -> &str { + MY_JOB_NAME + } + + async fn run(&self, args: &JobRunArgs) -> JobOutcome { + // …do work… + match self.do_the_work().await { + Ok(count) => JobOutcome::ok_with( + count, + serde_json::json!({ + // Anything useful for `GET /api/admin/jobs` or logs. + "bytes_processed": total_bytes, + "forced": args.force, + }), + ), + Err(e) => JobOutcome::err(format!("my_job failed: {e}")), + } + } +} +``` + +Rules: +- **`name()` must be stable** — it appears in log lines, admin URLs, + and `admin.background_runs.job_name` (once Part 2 lands). Renaming + is a breaking change to operator scripts and log dashboards. +- **`args.force` semantics are per-job.** If your job has no + acceleration mode (e.g. reconciliation, which is always idempotent), + ignore it. If it does (e.g. skip a grace window), document the + behaviour on the handler docstring. +- **Return `JobOutcome::ok_with(count, extra)`.** `count` is the + primary scalar operators read (rows swept, blobs reclaimed). + `extra` is a free-form JSON blob surfaced in the log line and the + admin listing. +- **Return `JobOutcome::err(msg)` on failure.** The supervisor logs + `outcome=err, cause=handler` and continues to the next tick. Don't + catch panics inside `run()` — the supervisor does it, and hiding + one loses the `cause=panicked` diagnostic. + +### 2. Add a `register()` method to your service + +Every service uses the chainable self-registration shape: + +```rust +impl MyService { + /// Register self with the periodic-job scheduler and return the + /// same `Arc` for DI-style chaining. + pub async fn register(self: Arc, registry: &JobRegistry) -> Arc { + let interval = /* Some(Duration::from_secs(...)) or None for on-demand */; + registry.register(self.clone(), interval, None /* no timeout */).await; + self + } +} +``` + +- **Interval `Some(dur)`** → **scheduled**. The supervisor fires the + handler every `dur`. +- **Interval `None`** → **on-demand only**. Never fires periodically; + only reachable via `POST /api/admin/jobs/{name}/trigger` or a + programmatic call to `registry.trigger(name, args)`. Use for jobs + whose periodic work happens elsewhere (dedup GC piggybacks on + trash cleanup) but which still benefit from a uniform admin trigger. +- **Timeout `Some(dur)`** → the supervisor wraps the handler in + `tokio::time::timeout`. Timeout trip is logged as + `outcome=err, cause=timeout`. Use sparingly — most native jobs don't + need it. Note: aborting a task mid-run is best-effort; a handler + that ignores await points may still finish in the background. + +### 3. Wire in `common/di.rs` + +One statement per service: + +```rust +let my_service = Arc::new(MyService::new(deps...)) + .register(&core.job_registry) + .await; +``` + +`register()` panics on error (duplicate job name = DI wiring bug; +boot must fail loud) and emits a `job.registered` log line on +success. No `if let Err(e)` scaffolding needed at the call site. + +`core.job_registry` is populated inside `create_core_services` and +lives on `CoreServices`. `SchedulerEngine` spawns the supervisor +task at the end of `build_app_state` after every service has +registered. + +### 4. That's it + +- `GET /api/admin/jobs` immediately lists the new job. +- `POST /api/admin/jobs/my_job/trigger` runs one dispatch off-schedule. +- `POST /api/admin/jobs/my_job/trigger?force=true` runs one dispatch + with `JobRunArgs { force: true }`. +- If the job is scheduled, the supervisor fires it at the configured + cadence. + +No handler registration in the router, no admin trigger endpoint to +add — the framework already covers those uniformly. + +--- + +## Admin surface + +Production admin endpoints, always on, audit-logged. Admin-only via +the standard `/api/admin/*` middleware (no dedicated feature flag). + +``` +GET /api/admin/jobs + → [{ name, interval_ms?, next_run_at?, last_run_at?, last_outcome?, running }] + +POST /api/admin/jobs/{name}/trigger[?force=] + → 200 { ok: true, outcome: { outcome: "ok" | "err", count, extra?, message? } } + → 404 { error: "job not registered", name } +``` + +- `interval_ms` / `next_run_at` are absent for on-demand jobs + (`skip_serializing_if=None`). +- `last_run_at` / `last_outcome` are absent until the first run + completes. +- `running` is `true` iff the in-flight permit is currently held. +- Every trigger call emits a `target: "audit"` log line + (`event = "job.trigger"`) before dispatch. + +--- + +## Log lines + +Uniform target: `oxicloud::scheduler`. + +| Event | Fields | When | +|---|---|---| +| `scheduler.started` | — | Supervisor loop starts | +| `scheduler.ready` | `registered = N` | All services have registered | +| `job.registered` | `job`, `cadence` ("every 24 h" / "on-demand") | Each `register()` call | +| `job.run` | `job`, `outcome` (ok\|err), `cause?` (handler\|timeout\|panicked), `count`, `elapsed_ms`, `extra?`, `error?` | Every dispatch | +| `job.tick_skipped` | `job`, `interval_ms`, `running_for_ms` | Tick fires while the previous run is still going | +| `job.trigger` | `job`, `force` (audit channel) | Admin `POST /trigger` | + +`elapsed_ms` is the raw scalar in the structured field; the human +message renders it as `12ms` / `1.4s` / `4m30s` so `tail -f` operators +see the duration inline. + +--- + +## Testing + +The scheduler's own tests live in `src/infrastructure/scheduler/` +(`registry.rs::tests`, `engine.rs::tests`) and use dummy handlers. +No integration test infrastructure is needed to add a new service — +your service's normal unit tests cover the `run()` logic, and the +Hurl suite `tests/api/admin_jobs.hurl` covers the admin surface +generically. + +If your service has a test that needs to construct the type WITHOUT +registering with a scheduler (e.g. isolated unit tests), skip the +`.register(®).await` chain and use the bare `Arc::new(Service::new(...))`. + +--- + +## Non-goals + +- **Cross-job dependencies.** No `depends_on` — each job runs + independently. If you find yourself needing "job B runs after job A + completes", route the completion signal through a lifecycle hook + (`FileLifecycleHook`, `BlobLifecycleHook`), not through the scheduler. +- **Distributed scheduling.** Single-process only. If OxiCloud ever + runs multi-node, the pattern is `SELECT … FOR UPDATE SKIP LOCKED` + on a lease table — not this design. +- **Cron expressions.** Fixed intervals only. Real cron + (day-of-week/month, arbitrary times) can layer on top later via a + `next_run: Box` trait; nothing needs it today. +- **Backfill on startup.** If the process was down when a scheduled + tick was due, the missed tick is NOT caught up — the next tick fires + at its normal interval. + +--- + +## Related + +- Plan doc: [`docs/plan/job-registry.md`](../plan/job-registry.md) +- Long-running / resumable jobs (Part 2, not yet built): + [`docs/plan/job-registry.md#part-2--recoverable-run-engine`](../plan/job-registry.md#part-2--recoverable-run-engine) +- Consistency checks (a future Part 2 consumer): + [`docs/plan/consistency-check.md`](../plan/consistency-check.md) diff --git a/docs/plan/consistency-check.md b/docs/plan/consistency-check.md index b2b17fdc..7b864917 100644 --- a/docs/plan/consistency-check.md +++ b/docs/plan/consistency-check.md @@ -1,5 +1,25 @@ # Plan — Resumable consistency checks + `StatefulAdapter` contract +> ⚠️ **PARTIALLY SUPERSEDED (Ed 2026-07-28).** The current shipping +> design organises consistency checks **by the subject they iterate** +> (drives / folders / files / storage), NOT by concern (blob / thumbnail +> / used_bytes). Each `*_consistency` job is a direct +> `RecoverableJobHandler` impl on the Part 2 engine — no +> `ConsistencyCheck` trait, no `StatefulAdapter` supertrait, no per- +> subsystem check registry. Cursor = row PK of the iterated subject. +> +> **See instead:** +> - Memory: `project_consistency_jobs_landscape` — the current taxonomy. +> - `docs/plan/job-registry.md` Part 2 §Native tenants — updated table. +> - `docs/architecture/jobs.md` — implementor guide. +> +> Sections below discuss `BlobConsistencyCheck`, `ThumbnailConsistencyCheck`, +> `UsedBytesConsistencyCheck` etc. as separate impls of a +> `ConsistencyCheck` trait. That IS retired. Read those sections for +> the invariants (grace-window trap, cursor discipline, findings +> idempotency) — they still apply. Ignore the trait shapes / +> registration wiring — the Part 2 engine covers those uniformly. + ## Context OxiCloud persists state in several independent subsystems: content-addressable @@ -38,14 +58,14 @@ This plan lands: compiles without declaring its consistency contract. 3. An **educational surface** in trait doc-comments — decision axes (severity, direction, grace, cursor) and canonical-example pointers. -4. **Consistency-specific persistence** — `admin.consistency_findings`, +4. **Consistency-specific persistence** — `jobs.run_findings`, idempotent-on-`(run_id, kind, resource_id)`. 5. A **first check** — `BlobConsistencyCheck` (both directions, blob-keyed cursor, severity split). **Layer boundary — the runtime is not in this plan.** The resumable execution engine (cursor persistence, exclusivity, cancel protocol, -crash recovery, `admin.background_runs` schema, `JobStore`, +crash recovery, `jobs.recoverable_runs` schema, `JobStore`, `RunOutcome`, `run_or_resume`) lives in `docs/plan/job-registry.md` Part 2. This plan describes what `ConsistencyCheck` implementors write and how the check-specific bits (findings, severity, @@ -53,7 +73,7 @@ write and how the check-specific bits (findings, severity, **Order:** ships **after** the job-registry Part 2 engine lands. Consistency closes an operator-visibility gap today, but it depends -on Part 2's `RecoverableJob` + `JobStore` + `admin.background_runs` +on Part 2's `RecoverableJobHandler` + `JobStore` + `jobs.recoverable_runs` primitives — those come first. Once both are in, consistency runs are admin-triggered v1, becoming periodic-triggered when a `JobRegistry` (Part 1) tenant wraps `run_or_resume` for each @@ -157,11 +177,11 @@ Race matrix — missing direction: Nothing before "byte-exact whole-table snapshot verification" needs a quiescent server. Reserve `concurrent_safe() = false` for that one. -### Resumability — runs live in Part 2's `background_runs` +### Resumability — runs live in Part 2's `recoverable_runs` -Consistency runs are ordinary `RecoverableJob`s. The runtime plumbing +Consistency runs are ordinary `RecoverableJobHandler`s. The runtime plumbing — cursor persistence, exclusivity, cancel protocol, crash recovery, -`admin.background_runs` schema, `JobStore` trait, `RunOutcome`, +`jobs.recoverable_runs` schema, `JobStore` trait, `RunOutcome`, `run_or_resume` helper — lives in `docs/plan/job-registry.md` Part 2. This plan does not redefine any of it. @@ -172,10 +192,10 @@ This plan does not redefine any of it. `SELECT DISTINCT ON (job_name)` return the last run of every check alongside every other background job. - Consistency's per-check knobs — `grace_window_secs`, `batch_size`, - `concurrent_safe` — live inside `background_runs.params` JSONB at + `concurrent_safe` — live inside `recoverable_runs.params` JSONB at run-start time. The check reads them back via `serde_json::from_value(store.params()?)`. -- `background_runs.stats` accumulates `{"scanned_count": …, +- `recoverable_runs.stats` accumulates `{"scanned_count": …, "findings_this_run": …}`; readers call `(stats->>'scanned_count')::bigint`. @@ -183,9 +203,9 @@ The findings themselves are Layer C (this plan) — they don't generalise to storage-migration or reextract: ```sql -CREATE TABLE admin.consistency_findings ( +CREATE TABLE jobs.run_findings ( id UUID PRIMARY KEY, - run_id UUID NOT NULL REFERENCES admin.background_runs(id) ON DELETE CASCADE, + run_id UUID NOT NULL REFERENCES jobs.recoverable_runs(id) ON DELETE CASCADE, kind TEXT NOT NULL, -- OrphanBlob / MissingBlob / ... severity TEXT NOT NULL, -- DataLoss / Reclaimable / ... resource_id TEXT NOT NULL, @@ -193,15 +213,15 @@ CREATE TABLE admin.consistency_findings ( found_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE (run_id, kind, resource_id) -- idempotent re-scan on resume ); -CREATE INDEX ON admin.consistency_findings (run_id, severity); +CREATE INDEX ON jobs.run_findings (run_id, severity); ``` -FK on `background_runs.id` links a finding back to the run that +FK on `recoverable_runs.id` links a finding back to the run that produced it; `ON DELETE CASCADE` clears findings when their run row is pruned by a future retention job. -`admin.*` is a NEW schema, created by Part 2's migration — keep it -distinct from `auth.*` / `storage.*` so operational tables don't +`jobs.*` is a NEW schema, created by Part 2's migration — keep it +distinct from `auth.*` / `storage.*` / `admin.*` so operational tables don't pollute domain schemas. ### Non-obvious traps @@ -223,7 +243,7 @@ learned the hard way in similar systems: transitioned (was `MissingBlob`, blob has since landed → drop the finding, not the whole run). 4. **Cooperative cancellation ONLY.** Between batches, poll - `background_runs.status`. A `tokio::spawn` abort mid-batch leaks — + `recoverable_runs.status`. A `tokio::spawn` abort mid-batch leaks — cursor unpersisted, findings half-written. Cancel path writes `status='Paused'` + current cursor before returning. 5. **Crash recovery on boot.** Any `status='Running'` at server start = @@ -484,7 +504,7 @@ impl ConsistencyRegistry { ## Admin surface -Consistency runs are ordinary `RecoverableJob`s (see +Consistency runs are ordinary `RecoverableJobHandler`s (see `docs/plan/job-registry.md` Part 2), so most operator actions reach them through the shared scheduler surface: @@ -510,7 +530,7 @@ GET /api/admin/jobs/consistency_{name}/runs/{id} ``` Findings enrichment on `runs/{id}` is consistency-specific — read -from `admin.consistency_findings` and joined into the response. +from `jobs.run_findings` and joined into the response. Everything else is generic Part 2 behaviour. Production surface — always on, audit-logged. No feature-flag gate. @@ -528,22 +548,22 @@ Production surface — always on, audit-logged. No feature-flag gate. `src/infrastructure/services/consistency/mod.rs` - `ConsistencyRegistry` (data structure only). - `PgCheckStore` — impl of `CheckStore` reading/writing - `admin.background_runs` (filtered to `job_name LIKE 'consistency_%'`) - + `admin.consistency_findings`. + `jobs.recoverable_runs` (filtered to `job_name LIKE 'consistency_%'`) + + `jobs.run_findings`. - `run_check(check, cursor, store)` — the runner that calls `run_resumable`, applies timeout, records outcome. ### 2. Schema migration `migrations/YYYYMMDDHHMMSS_background_runs_admin_schema.sql` — creates -the merged `admin.background_runs` table shared with the JobRegistry -plan. Consistency checks own the `admin.consistency_findings` table -alone and reference `background_runs.id` via FK. +the merged `jobs.recoverable_runs` table shared with the JobRegistry +plan. Consistency checks own the `jobs.run_findings` table +alone and reference `recoverable_runs.id` via FK. ```sql CREATE SCHEMA IF NOT EXISTS admin; -CREATE TABLE admin.background_runs ( +CREATE TABLE jobs.recoverable_runs ( id UUID PRIMARY KEY, job_name TEXT NOT NULL, -- 'consistency_blobs', 'storage_migration', 'reextract_audio', ... status TEXT NOT NULL, -- Running / Paused / Completed / Failed / CancelRequested @@ -556,13 +576,13 @@ CREATE TABLE admin.background_runs ( error_message TEXT ); CREATE UNIQUE INDEX one_active_run_per_job - ON admin.background_runs (job_name) + ON jobs.recoverable_runs (job_name) WHERE status IN ('Running', 'Paused'); -CREATE INDEX ON admin.background_runs (last_progress_at) WHERE status = 'Running'; +CREATE INDEX ON jobs.recoverable_runs (last_progress_at) WHERE status = 'Running'; -CREATE TABLE admin.consistency_findings ( +CREATE TABLE jobs.run_findings ( id UUID PRIMARY KEY, - run_id UUID NOT NULL REFERENCES admin.background_runs(id) ON DELETE CASCADE, + run_id UUID NOT NULL REFERENCES jobs.recoverable_runs(id) ON DELETE CASCADE, kind TEXT NOT NULL, severity TEXT NOT NULL, resource_id TEXT NOT NULL, @@ -570,7 +590,7 @@ CREATE TABLE admin.consistency_findings ( found_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE (run_id, kind, resource_id) ); -CREATE INDEX ON admin.consistency_findings (run_id, severity); +CREATE INDEX ON jobs.run_findings (run_id, severity); ``` ### 3. Supertrait bounds on existing state-owning ports @@ -631,7 +651,7 @@ Ship this PR without the actual checks. Grep `TODO(consistency)` = punch list. `src/interfaces/api/handlers/admin_handler.rs` - `start_consistency_check(name, force)` — insert an - `admin.background_runs` row with `job_name = 'consistency_'` + `jobs.recoverable_runs` row with `job_name = 'consistency_'` and `status = 'Running'`, spawn a tokio task calling `run_check`, return `run_id`. Concurrent triggers hit the partial unique index and short-circuit to returning the surviving row. @@ -651,7 +671,7 @@ In `AppServiceFactory` init, after DB pool is up: ```rust sqlx::query!( - "UPDATE admin.background_runs + "UPDATE jobs.recoverable_runs SET status = 'Paused', error_message = COALESCE(error_message, 'server restart mid-run') WHERE job_name LIKE 'consistency_%' @@ -660,7 +680,7 @@ sqlx::query!( ``` Filtering on `job_name LIKE 'consistency_%'` scopes the sweep to -consistency runs; other tenants of `background_runs` (storage +consistency runs; other tenants of `recoverable_runs` (storage migration, reextract-*) run the same auto-Pause sweep from their own boot-time hook. The JobRegistry supervisor's boot check may generalise this into a single scheduler-wide sweep — until then, one diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 33dc21bd..0a82658e 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -115,8 +115,8 @@ are already there. Keeping it as its own loop is right. of trash-cleanup + storage-usage + db_pool_monitor + dedup GC + grant-cleanup + tree-etag flush + content-index. High mechanical payoff, zero new schema, minimal review surface. -2. **Part 2 lands next** — introduces `admin.background_runs` schema - + `RecoverableJob` trait + `JobStore` port + `run_or_resume`. On +2. **Part 2 lands next** — introduces `jobs.recoverable_runs` schema + + `RecoverableJobHandler` trait + `JobStore` port + `run_or_resume`. On its own PR (schema change deserves independent review). 3. **Consistency-check framework (`docs/plan/consistency-check.md`)** lands third, consuming Part 2 as its runtime. @@ -432,16 +432,16 @@ into the scheduler. ## Part 2 — Recoverable-Run Engine -### Contract — `RecoverableJob` trait +### Contract — `RecoverableJobHandler` trait Sibling to `JobHandler`, NOT a subtrait. A stateless job that only implements `JobHandler` never needs to know Part 2 exists. ```rust #[async_trait] -pub trait RecoverableJob: Send + Sync { +pub trait RecoverableJobHandler: Send + Sync { /// Stable snake_case identifier — matches the `job_name` column - /// in `admin.background_runs`. + /// in `jobs.recoverable_runs`. fn name(&self) -> &str; /// Long-running, cooperative scan. The store is the job's ONLY @@ -479,13 +479,13 @@ pub enum RunOutcome { ### `JobStore` trait The port the engine passes to a recoverable job. Backed by -`admin.background_runs` in production; can be mocked for unit tests. +`jobs.recoverable_runs` in production; can be mocked for unit tests. ```rust #[async_trait] pub trait JobStore: Send + Sync { /// The `run_id` this handler was invoked with. Uniquely identifies - /// the row in `admin.background_runs`. + /// the row in `jobs.recoverable_runs`. fn run_id(&self) -> Uuid; /// Fixed at run start; used by consistency checks (and any other @@ -517,12 +517,12 @@ instance) are separate traits the impl composes on top of `JobStore`. `JobStore` itself carries no findings/severity concept — those are Layer C in the consistency-check plan, not the engine's concern. -### Schema — `admin.background_runs` +### Schema — `jobs.recoverable_runs` ```sql CREATE SCHEMA IF NOT EXISTS admin; -CREATE TABLE admin.background_runs ( +CREATE TABLE jobs.recoverable_runs ( id UUID PRIMARY KEY, job_name TEXT NOT NULL, status TEXT NOT NULL, -- Running / Paused / CancelRequested / Completed / Failed @@ -536,10 +536,10 @@ CREATE TABLE admin.background_runs ( ); CREATE UNIQUE INDEX one_active_run_per_job - ON admin.background_runs (job_name) + ON jobs.recoverable_runs (job_name) WHERE status IN ('Running', 'Paused', 'CancelRequested'); -CREATE INDEX ON admin.background_runs (last_progress_at) +CREATE INDEX ON jobs.recoverable_runs (last_progress_at) WHERE status = 'Running'; ``` @@ -549,9 +549,9 @@ so it survives concurrent triggers, admin-vs-scheduler races, and transaction interleavings. The `CancelRequested` inclusion prevents a second trigger during cancel from spawning a parallel run. -`admin.*` is a NEW schema — kept distinct from `auth.*` / `storage.*` +`jobs.*` is a NEW schema — kept distinct from `auth.*` / `storage.*` / `admin.*` so operational tables don't pollute domain schemas. Consistency -checks own their own `admin.consistency_findings` in the same +checks own their own `jobs.run_findings` in the same schema. Cursor is `BYTEA`, not JSONB, because per-job cursors are fixed-shape @@ -579,7 +579,7 @@ One `UPDATE` per checkpoint. Cheap, no row-lock contention (this process owns the row): ```sql -UPDATE admin.background_runs +UPDATE jobs.recoverable_runs SET cursor = $2, stats = jsonb_set( stats, @@ -618,7 +618,7 @@ that's a success. Log lines stay meaningful (`outcome=ok`, The engine module exposes: ```rust -pub async fn run_or_resume( +pub async fn run_or_resume( job: Arc, store_factory: &dyn JobStoreFactory, ) -> JobOutcome @@ -671,7 +671,7 @@ At `AppServiceFactory` init, after DB pool is up: ```rust sqlx::query!( - "UPDATE admin.background_runs + "UPDATE jobs.recoverable_runs SET status = 'Paused', error_message = COALESCE(error_message, 'server restart mid-run') WHERE status IN ('Running', 'CancelRequested')" @@ -703,18 +703,28 @@ GET /api/admin/jobs/{name}/runs/{id} ### Native tenants (Part 2) -- **Blob storage backend migration.** `migration_job.rs` becomes a - `RecoverableJob` impl. Cursor = last processed blob hash. Retires - the `Arc>` in-memory struct. -- **Reextract audio metadata.** Currently synchronous inside the - admin HTTP request. Becomes a `RecoverableJob` iterating audio - files by `file_id`. -- **Reextract image/video capture dates.** Same as above. -- **Consistency-check runs.** Every `ConsistencyCheck` impl gets - wrapped by a `RecoverableJob` adapter; the wrapper writes to - `admin.background_runs` via `JobStore`, and separately writes - findings to `admin.consistency_findings` via a check-specific - extension trait. See `docs/plan/consistency-check.md`. +Consistency checks are organized **by the subject they iterate**, not +by the concern they check. Cursor = row PK of that subject. Adding a +new check = adding a per-row branch inside the job that walks that +subject. See memory `project_consistency_jobs_landscape` for the full +rationale + the merges/separations that fall out of the rule. + +| Tenant | Iterates | Cursor | v1 checks | Notes | +|---|---|---|---|---| +| `drives_consistency` | `storage.drives` | drive UUID | `used_bytes` drift (drive + user envelope) | Shipped Slice 3. | +| `folders_consistency` | `storage.folders` | folder UUID | `parent_trashed_mismatch` (live folder under trashed parent), `path_mismatch`, `lpath_mismatch` — both materialised columns compared to parent-chain reconstruction | Shipped Slice 4. Room to grow: `drive_id_parent_mismatch`, `orphan_root` (self-join already loads the fields). | +| `files_consistency` | `storage.files` | file UUID | `parent_folder_trashed` (live file under trashed folder), `missing_blob` (severity `data_loss` — `blob_hash` present in neither `storage.blobs` nor `storage.chunk_manifests`), `chunk_missing` (severity `data_loss` — manifest exists but points at chunks absent from `storage.blobs`; typical dedup GC race), `blob_size_mismatch` (denormalised `files.size` diverges from the authoritative size — manifest first, blob fallback) | Shipped Slice 6, CDC-aware Slice 10. Handles both storage paths: `storage.chunk_manifests` (post-Apr-2026 FastCDC ingest, dominant path) and `storage.blobs` (pre-CDC whole-file blob, legacy fallback). Physical backend-existence checks (chunk bytes actually on disk) belong in `storage_consistency`. Room to grow: `drive_id_parent_mismatch`, mime-type reconciliation. | +| `storage_consistency` | Storage backend (fs / S3) | object key / path | Each blob has a `storage.blobs` row (orphan detection) | `?deep=true` adds re-BLAKE3 + mime sniff. Orphan-side of the old bidirectional blob check + former `blob_integrity` + former `thumbnail_consistency`. | +| `grants_consistency` (future) | `storage.role_grants` | grant UUID | subject/resource/granted_by exist | | +| `storage_migration` | `storage.blobs` (source) → target backend | blob hash | Copy bytes; failures → `stats.failed_blobs` (and eventually `jobs.run_findings`) | Retires `Arc>` in `migration_job.rs`. | +| `reextract_audio` | `storage.files` where audio | file UUID | Re-run audio-tag parser, upsert `audio_metadata` | Retires synchronous admin-request execution. | +| `reextract_image` | `storage.files` where image/video | file UUID | Re-run EXIF/container date parser, upsert capture date | Same shape as reextract_audio. | +| `consistency_batch` (wrapper) | Iterates registered `*_consistency` jobs | — (JobHandler, not RecoverableJobHandler) | Sequentially triggers each sub-job; `?deep=true` propagates | Shipped Slice 5. One-click "run all" without per-job clicks; exclusivity via `job_name` prevents concurrent batches from stepping on each other. Batch itself always returns `Ok` — child failures land in `outcome.extra.per_check[].outcome`. | + +**Not consistency**: `POST /api/admin/dedup/recalculate` is aggregate- +stats-only (`unique_blobs`, `total_references`, `bytes_saved`) — one +SELECT + one UPDATE. Kept as its own admin endpoint; do NOT fold into +`storage_consistency` (different semantic — recompute vs verify). ### Verification (Part 2) @@ -733,7 +743,7 @@ GET /api/admin/jobs/{name}/runs/{id} 6. **Idempotent replay:** for consistency-check specifically, verify that re-processing the last unpersisted batch does NOT double-record findings (`UNIQUE (run_id, kind, resource_id)` on - `admin.consistency_findings`). + `jobs.run_findings`). 7. **`RunOutcome` bridge log lines:** completed run logs `outcome=ok, extra.completed=true`; paused logs `outcome=ok, extra.paused=true`; failed logs `outcome=err, cause=handler`. @@ -790,26 +800,154 @@ endpoint returns a uniform `{ ok, outcome: JobOutcome }` envelope with job-specific fields under `outcome.extra`. Any external caller reading the old fields needs updating. +### Admin UI — /admin/jobs page (frontend, future slice) + +Operators shouldn't have to `curl` these endpoints in production — +they need a UI. Ships as a SvelteKit route once the backend surface is +complete. Rough shape: + +**Route:** `/admin/jobs` (SvelteKit page under `frontend/src/routes/admin/jobs/`). +**Access:** admin-only; same guard as the rest of `/admin/*`. + +**Page layout — one table, one drawer:** + +``` +┌── Jobs ─────────────────────────────────────────────────────────────┐ +│ Name Cadence Last run Status Actions │ +│ ───────────────────────────────────────────────────────────────────│ +│ trash_cleanup every 24 h 3h ago ok [Run] │ +│ storage_reconcile every 10 m 4m ago ok [Run] │ +│ dedup_gc on-demand 1d ago ok [Run] │ +│ grant_cleanup every 24 h never — [Run] │ +│ drives_consistency on-demand never — [Run] │ +│ consistency_batch on-demand never — [Run] [Run deep] │ +└─────────────────────────────────────────────────────────────────────┘ +``` + +Row click opens a right-side drawer with: +- Full JSON of the last outcome (`extra` fields explained per-job). +- For recoverable jobs: run history table (`GET /jobs/{name}/runs`), + each row expandable to full `RunSummary` (cursor, stats, params, + error_message). +- Per-run actions: `Cancel` (for Running rows only), `Trigger resume` + (for Paused rows — same trigger endpoint, `run_or_resume` picks + up the cursor). + +**Data flow:** +- `GET /api/admin/jobs` — populates the main table. Polled every 5 s + when the page is visible (`document.visibilityState`). +- `POST /api/admin/jobs/{name}/trigger` — the "Run" button. `deep=true` + query for the "Run deep" variant (currently only shown on + `consistency_batch`). +- `POST /api/admin/jobs/{name}/cancel` — Cancel button on a Running + recoverable run. +- `GET /api/admin/jobs/{name}/runs` — populates the history table when + the drawer opens. +- `GET /api/admin/jobs/{name}/runs/{id}` — populates the per-run + detail expander. + +**No new backend endpoints required** — every screen is driven by +what already exists. + +**Visual conventions:** +- Status colour: `ok` = green, `err` = red, `Running` = blue-pulse, + `Paused` = amber, `CancelRequested` = amber-flash, `Completed` = + neutral grey, `Failed` = red. +- Findings surfacing is live as of Slice 7 (`jobs.run_findings` + + `store.record_finding` + `GET /api/admin/jobs/{name}/runs/{id}/findings`). + Drawer's "Findings" tab renders `kind`, `severity`, `resource_id`, + and per-tenant `detail` JSON. + +**Slice ordering:** frontend page is a follow-up PR, not blocking any +backend slice. Order of appearance: +1. Backend Part 2 slices (engine, admin surface, first tenant) — done. +2. `jobs.run_findings` table + `store.record_finding` API — done (Slice 7). +3. `consistency_batch` + more tenants — done (Slices 5–6: drives + folders + files, plus batch). +4. Frontend `/admin/jobs` page — takes the completed backend surface + as-is; no backend changes required by the UI landing. +5. Progress estimation on `RunSummary.progress` (`fraction`, `kind`, + `scanned`, `total`) — **done (Slice 9)**. Tenants that CAN count + their subject override `RecoverableJobHandler::count_total()`; + `run_or_resume` seeds `params.total_rows` + `params.progress_kind` + on fresh runs; `row_to_summary` derives the `progress` block at + serialisation time. UI renders a bar; `kind = "approximate"` runs + get a striped fill so operators recognise proxy-derived + estimates. See memory `project_job_progress_estimation`. + +### Notifications & alerting + +Silent failure is the enemy — a consistency check that finds a +data-loss finding at 3 AM Sunday should reach an operator, not sit in +the log stream unread. When SMTP is wired, the supervisor emits an +alert email on the following: + +- **Any job dispatch returns `JobOutcome::Err`.** Applies to both + Part 1 handler errors and Part 2 recoverable `RunOutcome::Failed` + (which translates to `Err` via `run_or_resume`'s bridge). Subject + line: `[OxiCloud] Job failed`. Body includes: job name, + cause (`handler|timeout|panicked`), error message, run_id (Part 2 + only), elapsed_ms, log-timestamp for grep, link to + `/admin/jobs?highlight=` when the UI lands. +- **Consistency check surfaces one or more findings** (deferred to + the `jobs.run_findings` migration). Applies only to `*_consistency` + tenants. Body includes: run_id, findings count grouped by + `(kind, severity)`, worst-severity example, link to + `/admin/jobs/{name}/runs/{id}` when the UI lands. + +**Delivery conditions:** +- Silent no-op when `email_sender` on `AppState` is `None` (SMTP not + configured). No error, no log spam — the mechanism is opt-in + through SMTP presence. +- Recipient: every user with `role = 'admin'`. Not a hardcoded + address — same rule as any admin-scoped notification the codebase + already sends. +- Rate limit: **at most 1 email per (job_name, kind) per 6 hours**, + keyed off an in-memory dedup table on `AppState`. Prevents a + flapping job (fails, retries, fails, ...) from mailbombing. + 6 h chosen to match the operator-attention interval — a real + ongoing failure gets 4 alerts/day, enough to be noticed, not + enough to be filtered. +- Configurable OFF per job via env: `OXICLOUD_JOB__ALERT_ON_FAIL=false` + (default `true`). Same shape as the existing enable/disable knobs. + +**Implementation notes** (for whichever slice picks this up): +- Reuses `EmailSender` port + `MagicLinkInviteService`-style templating + under `askama`. New template files: + `templates/emails/job_failed.{html,txt}` and + `templates/emails/consistency_findings.{html,txt}`. +- Dedup table lives on `AppState.job_alert_dedup: + Arc>>`. Cleaned lazily on + insert. +- Called from `SchedulerEngine::log_outcome` (Part 1 path) and from + `run_or_resume`'s terminal-write branch (Part 2 path). Both already + see the `JobOutcome`; adding a fire-and-forget email dispatch is + ~10 lines each. + +**Scope-out:** no Slack / webhook / PagerDuty integration in v1. +Email is the ONE alert channel until an operator concretely asks for +another. Layering webhooks on top later is trivial — same +"terminal outcome → notification" hook, different sink. + ### Config surface — env vars -Canonical form for every job (Part 1 or Part 2 alike, AND for core -workers even though they don't register with the scheduler): +**No new convention.** Each service keeps its natural per-service +prefix (`OXICLOUD_GRANT_CLEANUP_*`, `OXICLOUD_STORAGE_USAGE_*`, …). +The `GET /api/admin/jobs` endpoint already gives operators a runtime +view of every registered job's interval, so grepping env-var prefixes +is no longer the primary discovery path. -``` -OXICLOUD_JOB__ENABLED -OXICLOUD_JOB__INTERVAL_HOURS # or _INTERVAL_SECS for sub-hour cadences -OXICLOUD_JOB__... # e.g. _GRACE_HOURS, _BATCH_SIZE -``` +Earlier drafts proposed a uniform `OXICLOUD_JOB__INTERVAL_*` +convention, with legacy names as warned aliases. Killed 2026-07-28 +(Ed): normalising only the interval knob while leaving domain-specific +tunables (`GRACE_DAYS`, `BATCH_SIZE`, …) at the natural prefix creates +*intra-service* prefix drift — worse than the *cross-service* drift it +was meant to solve. A service either goes fully to `OXICLOUD_JOB_*` +(disruptive rename of every knob) or fully stays at its native prefix +(no rename). We stay. -Core workers reuse this naming purely for uniform operator ergonomics -(e.g. `OXICLOUD_JOB_TREE_ETAG_FLUSH_INTERVAL_MS`) — the convention is -what operators grep for; whether the loop is scheduler-driven or a -dedicated `tokio::spawn` is an implementation detail they don't see. - -Existing per-service env vars keep working as **aliases** during -migration — `OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS` reads first, falls -back to `OXICLOUD_JOB_GRANT_CLEANUP_INTERVAL_HOURS`. Deprecated aliases -warn once on startup and stay recognised through one minor version. +The one real gap is **trash_cleanup has no env var today** (hardcoded +24h in DI). Adding `OXICLOUD_TRASH_CLEANUP_INTERVAL_HOURS` when we +need it uses the natural prefix — no new convention needed. ### Logging schema @@ -874,7 +1012,7 @@ precludes it. ### Job-history observability -`admin.background_runs` already carries the latest run per Part 2 job +`jobs.recoverable_runs` already carries the latest run per Part 2 job — "last run time + status" is a `SELECT DISTINCT ON (job_name) …` query. Deeper history (retention window, per-run drill-down UI) is deferred; the log stream is the source of truth for older runs. @@ -889,7 +1027,7 @@ No such need today. - **Cross-job dependencies.** Register-time ordering only, not runtime graph. -- **Retention pruning of terminal `background_runs` rows.** Deferred +- **Retention pruning of terminal `recoverable_runs` rows.** Deferred until the volume warrants a policy. - **Prometheus / OpenMetrics export.** Log-only for now. - **Distributed scheduling.** Single-process. If OxiCloud ever runs diff --git a/frontend/src/lib/api/endpoints/adminJobs.ts b/frontend/src/lib/api/endpoints/adminJobs.ts new file mode 100644 index 00000000..fb7f4764 --- /dev/null +++ b/frontend/src/lib/api/endpoints/adminJobs.ts @@ -0,0 +1,171 @@ +/** + * Admin JobRegistry endpoints — `/api/admin/jobs*` (see + * `docs/plan/job-registry.md`). Powers the "Jobs" tab of the admin panel. + * + * Every mutation goes through the standard admin auth path (Bearer JWT + * + admin-middleware role check). Read endpoints are cheap enough to + * poll while the panel is open. + */ +import { apiFetch, apiJson } from '$lib/api/client'; +import { getCsrfHeaders } from '$lib/api/csrf'; +import type { Finding, JobOutcome, JobSummary, RunSummary } from '$lib/api/types'; + +const JSON_HEADERS = { 'Content-Type': 'application/json' }; + +/** + * Envelope wrapping the outcome from `POST /api/admin/jobs/{name}/trigger`. + * `ok: true` means "dispatch reached the handler"; the handler's own + * pass/fail is in `outcome.outcome`. For `consistency_batch`, per-child + * outcomes are inside `outcome.extra.per_check`. + */ +export interface TriggerResponse { + ok: boolean; + outcome: JobOutcome; +} + +/** Envelope from `POST /api/admin/jobs/{name}/cancel`. `run_id` is + * the id of the run whose `Running` status was flipped to + * `CancelRequested` (null when nothing was in flight to cancel). */ +export interface CancelResponse { + ok: boolean; + run_id: string | null; +} + +/** + * `GET /api/admin/jobs` — full registry snapshot. One row per registered + * job (periodic + recoverable + coordinators like `consistency_batch`, + * which register as plain JobHandlers). + */ +export function listJobs(): Promise { + return apiJson('/api/admin/jobs', { credentials: 'same-origin' }); +} + +/** + * `POST /api/admin/jobs/{name}/trigger?force=X&deep=X` — dispatch a job + * on-demand. `force` bypasses per-tenant idempotency checks (e.g. + * `trash_cleanup` skipping when nothing is due). `deep` opts into slow + * variants (currently only `storage_consistency`, propagated by + * `consistency_batch` to every child). + * + * Throws on 4xx / 5xx with the backend's error message when present. + * A 404 means the job name isn't registered — surface that specifically + * so callers can distinguish "typo" from "handler blew up". + */ +export async function triggerJob( + name: string, + opts: { force?: boolean; deep?: boolean } = {} +): Promise { + const params = new URLSearchParams(); + if (opts.force) params.set('force', 'true'); + if (opts.deep) params.set('deep', 'true'); + const q = params.toString(); + const url = `/api/admin/jobs/${encodeURIComponent(name)}/trigger${q ? `?${q}` : ''}`; + const res = await apiFetch(url, { + method: 'POST', + credentials: 'same-origin', + headers: { ...JSON_HEADERS, ...getCsrfHeaders() } + }); + if (!res.ok) { + let msg = `trigger failed: ${res.status}`; + try { + const body = (await res.json()) as { error?: string; message?: string }; + msg = body.error ?? body.message ?? msg; + } catch { + /* no JSON body */ + } + throw new Error(msg); + } + return (await res.json()) as TriggerResponse; +} + +/** + * `POST /api/admin/jobs/{name}/cancel` — cooperatively request cancel + * of the currently running instance. The handler observes it on its + * next `store.status()` poll and returns `RunOutcome::Paused` at the + * next safe boundary. If nothing is running, this is a no-op that + * returns `run_id: null`. + */ +export async function cancelJob(name: string): Promise { + const res = await apiFetch(`/api/admin/jobs/${encodeURIComponent(name)}/cancel`, { + method: 'POST', + credentials: 'same-origin', + headers: { ...JSON_HEADERS, ...getCsrfHeaders() } + }); + if (!res.ok) { + let msg = `cancel failed: ${res.status}`; + try { + const body = (await res.json()) as { error?: string; message?: string }; + msg = body.error ?? body.message ?? msg; + } catch { + /* no JSON body */ + } + throw new Error(msg); + } + return (await res.json()) as CancelResponse; +} + +/** + * `GET /api/admin/jobs/{name}/runs?limit=N` — history of recoverable + * runs for `name`, newest first. Backend caps `limit` at 100. + */ +export function listRuns(name: string, limit = 20): Promise { + return apiJson(`/api/admin/jobs/${encodeURIComponent(name)}/runs?limit=${limit}`, { + credentials: 'same-origin' + }); +} + +/** Envelope from `POST /api/admin/jobs/runs/purge`. `purged` is + * the count of terminal-run rows deleted (findings cascade with + * their parent run via the FK, no separate counter). */ +export interface PurgeResponse { + purged: number; + retention_days: number; +} + +/** + * `POST /api/admin/jobs/runs/purge?days=N` — operator-triggered + * retention cleanup. Deletes terminal runs (`Completed`, `Failed`) + * with `completed_at` older than `days` days ago; associated + * `jobs.run_findings` rows drop with them via CASCADE. Non-terminal + * runs (`Running`, `Paused`, `CancelRequested`) are ALWAYS + * preserved regardless of age. + * + * Backend enforces a minimum of 1 day defensively. + */ +export async function purgeJobRuns(days = 30): Promise { + const res = await apiFetch(`/api/admin/jobs/runs/purge?days=${days}`, { + method: 'POST', + credentials: 'same-origin', + headers: { ...JSON_HEADERS, ...getCsrfHeaders() } + }); + if (!res.ok) { + let msg = `purge failed: ${res.status}`; + try { + const body = (await res.json()) as { error?: string; message?: string }; + msg = body.error ?? body.message ?? msg; + } catch { + /* no JSON body */ + } + throw new Error(msg); + } + return (await res.json()) as PurgeResponse; +} + +/** + * `GET /api/admin/jobs/{name}/runs/{id}/findings?limit=N&offset=M` — + * paginated findings for a specific run. Empty list = clean run, + * 404 = unknown run id. + */ +export function listFindings( + name: string, + runId: string, + opts: { limit?: number; offset?: number } = {} +): Promise { + const params = new URLSearchParams(); + params.set('limit', String(opts.limit ?? 100)); + if (opts.offset) params.set('offset', String(opts.offset)); + return apiJson( + `/api/admin/jobs/${encodeURIComponent(name)}/runs/${encodeURIComponent(runId)}/findings?${params}`, + { credentials: 'same-origin' } + ); +} diff --git a/frontend/src/lib/api/types.ts b/frontend/src/lib/api/types.ts index 38da7e5b..c23efc17 100644 --- a/frontend/src/lib/api/types.ts +++ b/frontend/src/lib/api/types.ts @@ -522,3 +522,102 @@ export interface FolderAncestorsResponse { ancestors: FolderAncestor[]; access_source: AccessSource; } + +// ─── Job registry (Part 1 + Part 2) ──────────────────────────────────────── +// +// Maps `src/infrastructure/scheduler/*` DTOs 1:1. See +// `docs/plan/job-registry.md` for the backend contract; the shapes below +// are what the `/api/admin/jobs*` endpoints emit. + +/** + * `JobOutcome` — the uniform outcome the scheduler logs and stores for + * every job dispatch. Serialised with `#[serde(tag = "outcome")]` so the + * discriminant is the `outcome` field, not the object key. + */ +export type JobOutcome = + | { outcome: 'ok'; count: number; extra?: unknown } + | { outcome: 'err'; message: string }; + +/** + * `JobSummary` — one row per registered job in `GET /api/admin/jobs`. + * Cadence + last-run bookkeeping. `interval_ms` / `next_run_at` are + * `undefined` on on-demand jobs (serde skips `Option::None`). + */ +export interface JobSummary { + name: string; + interval_ms?: number; + next_run_at?: string; + last_run_at?: string; + last_outcome?: JobOutcome; + running: boolean; +} + +/** + * `RunStatus` values allowed in `jobs.recoverable_runs.status`. The + * non-terminal set (Running / Paused / CancelRequested) is what the + * DB's `one_active_run_per_job` partial unique index scopes. + */ +export type RunStatus = 'Running' | 'Paused' | 'CancelRequested' | 'Completed' | 'Failed'; + +/** + * `RunSummary` — one row per recoverable-job run from + * `GET /api/admin/jobs/{name}/runs`. Terminal + non-terminal rows both + * appear. `stats` / `params` are opaque JSON — job-specific shape; + * consumers should key off `job_name` to decide what to render. + * `cursor_hex` is present only when the run has advanced past the + * initial state (paused mid-scan is the typical case). + */ +export interface RunSummary { + id: string; + job_name: string; + status: RunStatus; + started_at: string; + last_progress_at: string; + completed_at?: string; + stats: Record; + params: Record; + cursor_hex?: string; + error_message?: string; + /** Populated when the tenant reported a countable subject at run + * start (`RecoverableJobHandler::count_total`). Absent when the + * tenant can't count — the UI hides the progress bar and falls + * back to raw `scanned_count`. */ + progress?: RunProgress; +} + +/** + * Confidence level of a `RunProgress` fraction. Wire lowercase per + * the `#[serde(rename_all = "lowercase")]` on the Rust enum. + * + * - `count` — `scanned_count / total_rows` where `total_rows` came + * from a definitive `COUNT(*)` on the subject table. + * - `approximate` — proxy-derived total (e.g. `storage_consistency` + * using DB blob count as a stand-in for backend object count). + * Fraction can legitimately exceed 1.0 at run end — the deviation + * quantifies the drift the check is looking for. + */ +export type ProgressKind = 'count' | 'approximate'; + +export interface RunProgress { + fraction: number; + kind: ProgressKind; + scanned: number; + total: number; +} + +/** + * `Finding` — one row from `GET /api/admin/jobs/{name}/runs/{id}/findings`. + * Persisted by consistency tenants via `store.record_finding()`. Consumers + * key off `kind` to know the shape of `detail` (per-tenant JSON — e.g. + * `stale_used_bytes` carries `{cached, actual, delta}`; `missing_blob` + * carries `{blob_hash}`; …). + */ +export interface Finding { + id: string; + run_id: string; + kind: string; + severity: string; + resource_id?: string; + detail: Record; + created_at: string; +} diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte new file mode 100644 index 00000000..0d55d949 --- /dev/null +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -0,0 +1,1380 @@ + + + +
+
+
+

{t('admin.jobs.title', 'Jobs')}

+

+ {t( + 'admin.jobs.hint', + 'Fires periodic + on-demand jobs. Consistency checks are safe to run at any time — they are read-only.' + )} +

+
+
+ {#if hasBatch} + + + {/if} + + +
+
+ + {#if loadError} +

{loadError}

+ {:else if !jobs} +

{t('common.loading', 'Loading…')}

+ {:else if jobs.length === 0} +

{t('admin.jobs.none_registered', 'No jobs registered.')}

+ {:else} + + + + + + + + + + + + + {#each jobs as job (job.name)} + {@const runs = runsByJob[job.name]} + {@const runsErr = runsErrorByJob[job.name]} + {@const runsLoading = runsLoadingByJob[job.name]} + {@const expandedRun = expandedRunByJob[job.name] ?? null} + {@const canExpand = isRecoverable(job)} + + + + + + + + + + {#if expandedJob === job.name} + + + + {/if} + {/each} + +
{t('admin.jobs.col_name', 'Name')}{t('admin.jobs.col_cadence', 'Cadence')}{t('admin.jobs.col_last_run', 'Last run')}{t('admin.jobs.col_outcome', 'Outcome')}{t('admin.jobs.col_state', 'State')}{t('admin.jobs.col_actions', 'Actions')}
+ {#if canExpand} + + {:else} + {job.name} + {/if} + {cadenceLabel(job)}{timeAgo(job.last_run_at)} +
+ {outcomeLabel(job)} + {#if actionableFindingCount(job) > 0} + {@const findings = actionableFindingCount(job)} + + {t('admin.jobs.n_findings', { n: findings }, '{{n}} findings')} + + {/if} + {#if anomalyFindingCount(job) > 0} + {@const notices = anomalyFindingCount(job)} + + {t('admin.jobs.n_notices', { n: notices }, '{{n}} notices')} + + {/if} +
+ 		+ {#if isRunning(job)} + + {t('admin.jobs.state_running', 'running')} + + {:else} + — + {/if} + + + {#if supportsDeep(job.name)} + + {/if} + {#if isRunning(job) && canExpand} + + {/if} +
+
+
+

{t('admin.jobs.runs_title', 'Recent runs')}

+ +
+ {#if runsErr} +

{runsErr}

+ {:else if !runs} +

{t('common.loading', 'Loading…')}

+ {:else if runs.length === 0} +

+ {t('admin.jobs.no_runs', 'No runs yet.')} +

+ {:else} + + + + + + + + + + + + + + {#each runs as run (run.id)} + {@const scanned = statNumber(run, 'scanned_count')} + {@const findingCount = statNumber(run, 'finding_count')} + {@const isRunExpanded = expandedRun === run.id} + + + + + + + + + + {#if isRunExpanded} + {@const findings = findingsByRun[run.id]} + {@const findingsErr = findingsErrorByRun[run.id]} + {@const fLoading = findingsLoadingByRun[run.id]} + + + + {/if} + {/each} + +
{t('admin.jobs.col_started_at', 'Started')}{t('admin.jobs.col_status', 'Status')}{t('admin.jobs.col_duration', 'Duration')} + {t('admin.jobs.col_progress', 'Progress')} + {t('admin.jobs.col_findings', 'Findings')} + {t('admin.jobs.col_error', 'Error')} +
+ + + {timeAgo(run.started_at)} + + + {run.status} + + + {runDurationLabel(run)} + + {#if run.progress} + {@const barPct = Math.min( + 100, + Math.max(0, run.progress.fraction * 100) + )} + {@const pctLabel = (run.progress.fraction * 100).toFixed(1) + '%'} +
+
+
+
+ + {run.progress.scanned}/{run.progress.total} + +
+ {:else if scanned != null} + + {t( + 'admin.jobs.progress_scanned_only', + { n: scanned }, + '{{n}} scanned' + )} + + {:else} + — + {/if} + 		+ {#if findingCount && findingCount > 0} + + {findingCount} + + {:else} + 0 + {/if} + + {#if run.error_message} + {run.error_message} + {:else} + — + {/if} +
+
+
+ + {t('admin.jobs.run_json', 'Run summary (JSON)')} + + 
{JSON.stringify(
+																				{
+																					id: run.id,
+																					status: run.status,
+																					started_at: run.started_at,
+																					last_progress_at: run.last_progress_at,
+																					completed_at: run.completed_at,
+																					stats: run.stats,
+																					params: run.params,
+																					cursor_hex: run.cursor_hex,
+																					error_message: run.error_message
+																				},
+																				null,
+																				2
+																			)}
+
+
+
+

+ {t('admin.jobs.findings_title', 'Findings')} +

+ +
+ {#if findingsErr} +

+ {findingsErr} +

+ {:else if !findings} +

+ {t('common.loading', 'Loading…')} +

+ {:else if findings.length === 0} +

+ {t('admin.jobs.no_findings', 'No findings — clean run.')} +

+ {:else} + + + + + + + + + + + {#each findings as f (f.id)} + {@const detail = (f.detail ?? {}) as Record< + string, + unknown + >} + {@const label = + (detail.path as string | undefined) ?? + (detail.name as string | undefined) ?? + null} + + + + + + + {/each} + +
+ {t('admin.jobs.col_kind', 'Kind')} + + {t('admin.jobs.col_severity', 'Severity')} + + {t('admin.jobs.col_resource', 'Resource')} + + {t('admin.jobs.col_detail', 'Detail')} +
{f.kind} + + {f.severity} + + + {#if label} +
+ {label} + {#if f.resource_id} + {f.resource_id} + {/if} +
+ {:else} + {f.resource_id ?? '—'} + {/if} + 		+ {JSON.stringify(f.detail)} +
+ {/if} +
+
+
+ {/if} +
+
+ {/if} +
+ + + + {#if purgeModal} +
{ + e.preventDefault(); + void confirmPurge(); + }} + > +

+ {t( + 'admin.jobs.purge_body', + 'Delete completed and failed run history older than the chosen number of days. Findings drop with their parent runs. Non-terminal runs (running, paused, cancel-requested) are always preserved.' + )} +

+ +
+ + +
+
+ {/if} + + + diff --git a/frontend/src/lib/components/AppShell.svelte b/frontend/src/lib/components/AppShell.svelte index 861f8ab2..934610e8 100644 --- a/frontend/src/lib/components/AppShell.svelte +++ b/frontend/src/lib/components/AppShell.svelte @@ -32,15 +32,12 @@ const palette = lazyComponent(() => import('$lib/components/CommandPalette.svelte')); interface NavLink { - href: - | '/files' - | '/shared' - | '/shared-with-me' - | '/recent' - | '/favorites' - | '/photos' - | '/music' - | '/trash'; + /** + * String rather than a literal union so admin links (which + * include a dynamic path segment) can share the same shape. + * `resolve()` accepts any string, so no type-level cost. + */ + href: string; label: string; icon: string; /** Stable key driving the per-section icon colour (see sidebar.css). */ @@ -69,12 +66,103 @@ { href: '/trash', label: t('nav.trash', 'Trash'), icon: 'trash', section: 'trash' } ]; + // Admin sidebar — populated when the URL is under /admin. The + // admin +page.svelte used to render its own horizontal tab + // strip; that was displaced here so the section navigation + // scales past ~7 items and matches deep-link URLs from the + // address bar. + const ADMIN_LINKS: NavLink[] = [ + { + href: '/admin', + label: t('admin.dashboard', 'Dashboard'), + icon: 'chart-pie', + section: 'admin-dashboard' + }, + { + href: '/admin/users', + label: t('admin.users', 'Users'), + icon: 'users', + section: 'admin-users' + }, + { + href: '/admin/drives', + label: t('admin.drives', 'Drives'), + icon: 'folder', + section: 'admin-drives' + }, + { + href: '/admin/mounts', + label: t('admin.mounts', 'External Mounts'), + icon: 'folder', + section: 'admin-mounts' + }, + { + href: '/admin/oidc', + label: t('admin.oidc', 'OIDC / SSO'), + icon: 'key', + section: 'admin-oidc' + }, + { + href: '/admin/storage', + label: t('admin.storage_tab', 'Storage'), + icon: 'database', + section: 'admin-storage' + }, + { + href: '/admin/smtp', + label: t('admin.smtp', 'Email (SMTP)'), + icon: 'envelope', + section: 'admin-smtp' + }, + { + href: '/admin/plugins', + label: t('admin.plugins', 'Plugins'), + icon: 'layer-group', + section: 'admin-plugins' + }, + { + href: '/admin/jobs', + label: t('admin.jobs.tab', 'Jobs'), + icon: 'cogs', + section: 'admin-jobs' + } + ]; + const isAdmin = $derived(session.user?.role === 'admin'); + // Any URL under /admin swaps the sidebar to admin mode. Uses + // startsWith so a trailing slash / query params / hash don't + // desync. Root `/admin` counts too (dashboard). + const isAdminSection = $derived(page.url.pathname.startsWith('/admin')); + const currentLinks = $derived(isAdminSection ? ADMIN_LINKS : LINKS); + function active(href: string): boolean { return page.url.pathname === href || page.url.pathname.startsWith(`${href}/`); } + // Sidebar-item active check. Non-admin links use `active()` + // (matches href + any subpath). Admin links need a stricter + // rule for `/admin` itself — a plain `startsWith('/admin/')` + // would light up the Dashboard item on `/admin/drives` too. + // So `/admin` matches ONLY the exact path; every other admin + // item uses the same startsWith rule as before. + function activeLink(href: string): boolean { + if (href === '/admin') return page.url.pathname === '/admin'; + return active(href); + } + + /** + * Data-driven sidebar links (`LINKS`, `ADMIN_LINKS`) hold + * runtime strings, not compile-time route keys. SvelteKit's + * typed `resolve()` refuses them; we know they're valid + * routes at runtime. Cast at the callsite, one place, so + * the template stays clean. + */ + function navHref(href: string): string { + // @ts-expect-error runtime-known route string, not a literal typed key + return resolve(href); + } + // ── Sidebar drop targets ───────────────────────────────────────────────── // The row-drag on `/files` (and other resource surfaces) sets a // `application/x-oxi-item` MIME with a JSON array of `{ id, name, kind }`. @@ -467,16 +555,50 @@
OxiCloud
-