feat(server): one-at-a-time storage backfill jobs + the storage events bridge

server/src/nest/storage/storage-events.service.ts

@@ -0,0 +1,27 @@
+import { Injectable } from '@nestjs/common';
+import type { ReplicaFailure } from './drivers/mirror.driver';
+
+/**
+ * In-process pub/sub for storage health events. Exists so consumers OUTSIDE
+ * StorageModule (the notifications notifier) can observe replica failures
+ * without StorageModule importing NotificationsModule — that would cycle
+ * (Notifications → Auth → Storage).
+ */
+@Injectable()
+export class StorageEventsService {
+  private readonly replicaFailureSubscribers: Array<(failure: ReplicaFailure) => void> = [];
+
+  onReplicaFailure(subscriber: (failure: ReplicaFailure) => void): void {
+    this.replicaFailureSubscribers.push(subscriber);
+  }
+
+  emitReplicaFailure(failure: ReplicaFailure): void {
+    for (const subscriber of this.replicaFailureSubscribers) {
+      try {
+        subscriber(failure);
+      } catch {
+        // A broken subscriber must never break the write path.
+      }
+    }
+  }
+}

server/src/nest/storage/storage-jobs.service.ts

@@ -0,0 +1,111 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { STORAGE_CATEGORIES, type StorageBackfillStatus } from '@trek/shared';
+import { MirrorDriver } from './drivers/mirror.driver';
+import { StorageRegistryService } from './storage-registry.service';
+
+export class BackfillTargetError extends Error {}
+export class BackfillBusyError extends Error {}
+
+interface ActiveJob {
+  status: StorageBackfillStatus;
+  cancelled: boolean;
+}
+
+/**
+ * One-at-a-time backfill registry (the ImportJobsService idiom, minus WS).
+ * Finished statuses linger for ttlMs so a remounted panel can show the
+ * outcome; the client polls the admin state while anything is running.
+ */
+@Injectable()
+export class StorageJobsService {
+  private readonly logger = new Logger(StorageJobsService.name);
+  private readonly jobs = new Map<string, ActiveJob>();
+  private ttlMs = 10 * 60_000;
+
+  constructor(private readonly registry: StorageRegistryService) {}
+
+  /** Test-only factory for a short TTL, since a second constructor param would confuse Nest DI. */
+  static withTtl(registry: StorageRegistryService, ttlMs: number): StorageJobsService {
+    const svc = new StorageJobsService(registry);
+    svc.ttlMs = ttlMs;
+    return svc;
+  }
+
+  startBackfill(mirrorName: string): void {
+    if ([...this.jobs.values()].some((job) => job.status.status === 'running')) {
+      throw new BackfillBusyError('a sync is already running — one backfill at a time');
+    }
+    // Scope rule: the categories whose EFFECTIVE backend is this mirror.
+    const snapshot = this.registry.snapshot();
+    const categories = STORAGE_CATEGORIES.filter(
+      (category) => snapshot.categories[category]?.backend === mirrorName,
+    );
+    if (categories.length === 0) {
+      throw new BackfillTargetError(`'${mirrorName}' is not a mirror routed by any category`);
+    }
+    const resolved = categories.map((category) => this.registry.resolve(category));
+    const driver = resolved[0]!.driver;
+    if (!(driver instanceof MirrorDriver)) {
+      throw new BackfillTargetError(`'${mirrorName}' is not a mirror backend`);
+    }
+    const prefixes = resolved.map((r) => r.keyPrefix);
+
+    const job: ActiveJob = {
+      cancelled: false,
+      status: {
+        backend: mirrorName,
+        status: 'running',
+        done: 0,
+        total: 0,
+        copied: 0,
+        skipped: 0,
+        failed: 0,
+        startedAt: Date.now(),
+      },
+    };
+    this.jobs.set(mirrorName, job);
+
+    // Detached: the driver instances are resolved above, so a registry
+    // reload() mid-run keeps this job on them (the in-flight guarantee).
+    void driver
+      .backfill(prefixes, {
+        onProgress: (progress) => {
+          job.status = { ...job.status, ...progress };
+        },
+        isCancelled: () => job.cancelled,
+      })
+      .then((result) => {
+        job.status = {
+          ...job.status,
+          ...result,
+          status: result.cancelled ? 'cancelled' : 'done',
+          finishedAt: Date.now(),
+        };
+      })
+      .catch((err: unknown) => {
+        job.status = {
+          ...job.status,
+          status: 'error',
+          error: err instanceof Error ? err.message : String(err),
+          finishedAt: Date.now(),
+        };
+        this.logger.error(`backfill '${mirrorName}' aborted: ${job.status.error}`);
+      })
+      .finally(() => {
+        setTimeout(() => {
+          if (this.jobs.get(mirrorName) === job && job.status.status !== 'running') this.jobs.delete(mirrorName);
+        }, this.ttlMs).unref?.();
+      });
+  }
+
+  cancelBackfill(mirrorName: string): boolean {
+    const job = this.jobs.get(mirrorName);
+    if (!job || job.status.status !== 'running') return false;
+    job.cancelled = true;
+    return true;
+  }
+
+  statuses(): StorageBackfillStatus[] {
+    return [...this.jobs.values()].map((job) => ({ ...job.status }));
+  }
+}

server/src/nest/storage/storage-registry.service.ts

@@ -7,6 +7,7 @@ import { decrypt_api_key } from '../common/crypto/apiKeyCrypto';
 import { LocalDriver } from './drivers/local.driver';
 import { MirrorDriver, type ReplicaFailure } from './drivers/mirror.driver';
 import { S3Driver } from './drivers/s3.driver';
+import { StorageEventsService } from './storage-events.service';
 import { DEFAULT_BACKUPS_ROOT, DEFAULT_UPLOADS_ROOT, GLOBAL_TEMP_DIR, SEED_CONFIG_PATH } from './storage-paths';
 import {
   assertNoMaskSentinels,
@@ -121,6 +122,7 @@ export class StorageRegistryService implements OnModuleInit {
   constructor(
     private readonly db: DatabaseService,
     private readonly env: RuntimeEnvService,
+    private readonly events: StorageEventsService,
   ) {}
 
   onModuleInit(): void {
@@ -173,6 +175,7 @@ export class StorageRegistryService implements OnModuleInit {
     if (this.failures.length > REPLICA_FAILURE_RING_SIZE) {
       this.failures = this.failures.slice(-REPLICA_FAILURE_RING_SIZE);
     }
+    this.events.emitReplicaFailure(failure);
   }
 
   replicaFailures(): readonly ReplicaFailure[] {

server/src/nest/storage/storage.module.ts

@@ -1,6 +1,8 @@
 import { Module } from '@nestjs/common';
 import { AppConfigModule } from '../app-config/app-config.module';
 import { AuditModule } from '../audit/audit.module';
+import { StorageEventsService } from './storage-events.service';
+import { StorageJobsService } from './storage-jobs.service';
 import { StorageRegistryService } from './storage-registry.service';
 import { StorageService } from './storage.service';
 import { StorageAdminService } from './storage-admin.service';
@@ -22,7 +24,7 @@ import { StorageAdminController } from './storage-admin.controller';
 @Module({
   imports: [AppConfigModule, AuditModule],
   controllers: [StorageAdminController],
-  providers: [StorageRegistryService, StorageService, StorageAdminService],
-  exports: [StorageService],
+  providers: [StorageRegistryService, StorageService, StorageAdminService, StorageEventsService, StorageJobsService],
+  exports: [StorageService, StorageEventsService],
 })
 export class StorageModule {}

server/tests/unit/nest/storage/storage-admin.service.test.ts

@@ -20,6 +20,7 @@ import { DatabaseService } from '../../../../src/nest/database/database.service'
 import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service';
 import { encrypt_api_key } from '../../../../src/nest/common/crypto/apiKeyCrypto';
 import { StorageAdminService } from '../../../../src/nest/storage/storage-admin.service';
+import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service';
 import { StorageRegistryService, BACKENDS_KEY, CATEGORIES_KEY } from '../../../../src/nest/storage/storage-registry.service';
 import { StorageService } from '../../../../src/nest/storage/storage.service';
 
@@ -62,7 +63,7 @@ function makeService(opts: { encryptionKeySet?: boolean; uploadsRoot?: string }
   const env = {
     env: () => ({ paths: {}, security: { encryptionKeySet: opts.encryptionKeySet ?? true } }),
   } as unknown as RuntimeEnvService;
-  const registry = new StorageRegistryService(db, env);
+  const registry = new StorageRegistryService(db, env, new StorageEventsService());
   registry.onModuleInit();
   const service = new StorageAdminService(db, registry, new StorageService(registry), env);
   return { service, registry, uploadsRoot };

server/tests/unit/nest/storage/storage-jobs.service.test.ts

@@ -0,0 +1,136 @@
+import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from 'vitest';
+
+const { testDb, dbMock } = vi.hoisted(() => {
+  const Database = require('better-sqlite3');
+  const db = new Database(':memory:');
+  db.exec('PRAGMA journal_mode = WAL');
+  db.exec('PRAGMA foreign_keys = ON');
+  return { testDb: db, dbMock: { db, closeDb: () => {}, reinitialize: () => {} } };
+});
+vi.mock('../../../../src/db/database', () => dbMock);
+vi.mock('../../../../src/config', () => ({ ENCRYPTION_KEY: 'storage-jobs-test-key' }));
+
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { Readable } from 'node:stream';
+import { createTables } from '../../../../src/db/schema';
+import { runMigrations } from '../../../../src/db/migrations';
+import { DatabaseService } from '../../../../src/nest/database/database.service';
+import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service';
+import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service';
+import { StorageRegistryService } from '../../../../src/nest/storage/storage-registry.service';
+import { StorageService } from '../../../../src/nest/storage/storage.service';
+import {
+  BackfillBusyError,
+  BackfillTargetError,
+  StorageJobsService,
+} from '../../../../src/nest/storage/storage-jobs.service';
+
+const db = new DatabaseService(testDb);
+
+beforeAll(() => {
+  createTables(testDb);
+  runMigrations(testDb);
+});
+
+const tmpDirs: string[] = [];
+function makeTmpDir(): string {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'trek-jobs-'));
+  tmpDirs.push(dir);
+  return dir;
+}
+function setSetting(key: string, value: string): void {
+  testDb.prepare('INSERT OR REPLACE INTO app_settings (key, value) VALUES (?, ?)').run(key, value);
+}
+beforeEach(() => {
+  testDb.prepare("DELETE FROM app_settings WHERE key LIKE 'storage.%'").run();
+});
+afterEach(() => {
+  vi.restoreAllMocks();
+  while (tmpDirs.length) fs.rmSync(tmpDirs.pop()!, { recursive: true, force: true });
+});
+
+/** A registry whose backups category routes through mirror 'm' (nas replica). */
+function makeWorld() {
+  const uploadsRoot = makeTmpDir();
+  const backupsRoot = makeTmpDir();
+  const nasRoot = makeTmpDir();
+  setSetting(
+    'storage.backends',
+    JSON.stringify([
+      { name: 'uploads-local', type: 'local', options: { root: uploadsRoot } },
+      { name: 'backups-local', type: 'local', options: { root: backupsRoot } },
+      { name: 'nas', type: 'local', options: { root: nasRoot } },
+      { name: 'm', type: 'mirror', options: { primary: 'backups-local', replicas: ['nas'] } },
+    ]),
+  );
+  setSetting('storage.categories', JSON.stringify({ backups: 'm' }));
+  const env = { env: () => ({ paths: {}, security: { encryptionKeySet: true } }) } as unknown as RuntimeEnvService;
+  const registry = new StorageRegistryService(db, env, new StorageEventsService());
+  registry.onModuleInit();
+  const storage = new StorageService(registry);
+  const jobs = new StorageJobsService(registry);
+  return { registry, storage, jobs, backupsRoot, nasRoot };
+}
+
+async function waitFor(predicate: () => boolean, ms = 5000): Promise<void> {
+  const start = Date.now();
+  while (!predicate()) {
+    if (Date.now() - start > ms) throw new Error('timed out waiting for job');
+    await new Promise((r) => setTimeout(r, 20));
+  }
+}
+
+describe('StorageJobsService', () => {
+  it('JOBS-001 backfills the categories routed through the mirror and lands on done', async () => {
+    const { storage, jobs, nasRoot } = makeWorld();
+    await storage.put('backups', 'old-backup.zip', Readable.from('zipzip'));
+    fs.rmSync(path.join(nasRoot, 'old-backup.zip'), { force: true }); // simulate pre-mirror object
+    jobs.startBackfill('m');
+    await waitFor(() => jobs.statuses().some((s) => s.backend === 'm' && s.status === 'done'));
+    const status = jobs.statuses().find((s) => s.backend === 'm')!;
+    expect(status).toMatchObject({ status: 'done', total: 1, done: 1, copied: 1, failed: 0 });
+    expect(fs.existsSync(path.join(nasRoot, 'old-backup.zip'))).toBe(true);
+  });
+
+  it('JOBS-002 rejects an unrouted or non-mirror name with BackfillTargetError', () => {
+    const { jobs } = makeWorld();
+    expect(() => jobs.startBackfill('nas')).toThrow(BackfillTargetError);
+    expect(() => jobs.startBackfill('ghost')).toThrow(BackfillTargetError);
+  });
+
+  it('JOBS-003 a second start while one runs throws BackfillBusyError (global, either backend)', async () => {
+    const { storage, jobs } = makeWorld();
+    for (let i = 0; i < 20; i++) await storage.put('backups', `b${i}.zip`, Readable.from('x'.repeat(1000)));
+    jobs.startBackfill('m');
+    expect(() => jobs.startBackfill('m')).toThrow(BackfillBusyError);
+    await waitFor(() => jobs.statuses().some((s) => s.status !== 'running'));
+  });
+
+  it('JOBS-004 cancel flips a running job to cancelled; cancelling a finished/unknown one returns false', async () => {
+    const { storage, jobs } = makeWorld();
+    for (let i = 0; i < 50; i++) await storage.put('backups', `c${i}.zip`, Readable.from('y'.repeat(2000)));
+    jobs.startBackfill('m');
+    expect(jobs.cancelBackfill('m')).toBe(true);
+    await waitFor(() => jobs.statuses().some((s) => s.backend === 'm' && s.status !== 'running'));
+    expect(jobs.statuses().find((s) => s.backend === 'm')!.status).toBe('cancelled');
+    expect(jobs.cancelBackfill('m')).toBe(false);
+    expect(jobs.cancelBackfill('ghost')).toBe(false);
+  });
+
+  it('JOBS-005 finished statuses expire after the TTL', async () => {
+    vi.useFakeTimers();
+    try {
+      const { storage, jobs } = makeWorld();
+      await storage.put('backups', 'one.zip', Readable.from('z'));
+      jobs.startBackfill('m');
+      // Drain the real async job under fake timers by flushing microtasks.
+      await vi.waitFor(() => expect(jobs.statuses()[0]?.status).toBe('done'));
+      vi.advanceTimersByTime(10 * 60_000 + 1000);
+      expect(jobs.statuses()).toEqual([]);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});

server/tests/unit/nest/storage/storage-registry.service.test.ts

@@ -25,6 +25,7 @@ import { runMigrations } from '../../../../src/db/migrations';
 import { DatabaseService } from '../../../../src/nest/database/database.service';
 import type { RuntimeEnvService } from '../../../../src/nest/app-config/runtime-env.service';
 import { encrypt_api_key } from '../../../../src/nest/common/crypto/apiKeyCrypto';
+import { StorageEventsService } from '../../../../src/nest/storage/storage-events.service';
 import { StorageRegistryService } from '../../../../src/nest/storage/storage-registry.service';
 import { LocalDriver } from '../../../../src/nest/storage/drivers/local.driver';
 import { MirrorDriver, type ReplicaFailure } from '../../../../src/nest/storage/drivers/mirror.driver';
@@ -99,7 +100,7 @@ function makeRegistry(opts: RegistryOpts = {}) {
     { placePhotoDir: opts.placePhotoDir },
     { encryptionKeySet: opts.encryptionKeySet ?? true },
   );
-  const registry = new StorageRegistryService(db, stub.env);
+  const registry = new StorageRegistryService(db, stub.env, new StorageEventsService());
   if (opts.boot !== false) registry.onModuleInit();
   return { registry, uploadsRoot, setUploadsRoot: (root: string) => rewriteUploadsOverride(root) };
 }
@@ -479,7 +480,7 @@ describe('seed-once storage-config.json import', () => {
   /** A registry with NO storage.* rows (makeRegistry seeds an override row, so build raw). */
   function makeUnseededRegistry(opts: { encryptionKeySet?: boolean } = {}) {
     const stub = makeEnvStub({}, { encryptionKeySet: opts.encryptionKeySet ?? true });
-    return new StorageRegistryService(db, stub.env);
+    return new StorageRegistryService(db, stub.env, new StorageEventsService());
   }
 
   function readRow(key: string): string | undefined {