perf,feat(events),ops: third optimization pass

Overview cold start:
- InstanceResourceService persists every live fetch into status_snapshots
  (category 'resources') and serves a snapshot younger than 2 minutes
  before calling upstream, so the first fleet overview after a restart
  costs zero device requests; instance deletion cleans up via FK cascade

Bounded retention for the four fastest-growing tables (15-minute sweep):
- connection_logs: one row per probe per beat had no automatic cleanup
- audit_events: new pruneBefore (90d)
- operation_preparations: every prepare/retry attempt inserted a row,
  terminal rows now expire after 7 days
- notification_queue: terminal rows pruned after 30 days

SSE events now cover instance lifecycle:
- create/update emit instance envelopes from the routes; the two-phase
  delete emits a job envelope on every terminal transition plus an
  instance envelope when the node actually disappears, so fleet and
  instance views invalidate in real time

Linux ops:
- 'install.sh unit' writes systemd user units for API and gateway with
  Restart=on-failure, 0600 secret injection, and
  MULTI_SIMADMIN_SYSTEMD_UNIT wired so console self-update restarts via
  systemctl -- closing the self-update loop on Linux

apps/api/src/application/audit/audit-query-service.ts

@@ -139,6 +139,17 @@ function parameterSummary(value: unknown): readonly RedactedParameterSummaryItem
 export class AuditQueryService {
   constructor(private readonly db: Database.Database) {}
 
+  /** Retention hook: drops audit records older than the cutoff. */
+  pruneBefore(cutoffIso: string): number {
+    if (typeof cutoffIso !== 'string' || !Number.isFinite(Date.parse(cutoffIso)))
+      throw new RangeError('cutoffIso must be a valid date-time');
+    return Number(
+      this.db
+        .prepare('DELETE FROM audit_events WHERE created_at < ?')
+        .run(new Date(cutoffIso).toISOString()).changes,
+    );
+  }
+
   summary(limit = 5): ControlPlaneAuditSummary {
     if (!Number.isSafeInteger(limit) || limit < 1 || limit > 20) validation('limit is invalid');
     const totalRow = this.db.prepare('SELECT COUNT(*) AS total FROM audit_events').get() as

apps/api/src/application/notifications/central-notification-service.ts

@@ -1248,6 +1248,20 @@ export class CentralNotificationService {
     if (result.changes !== 1) throw new CentralNotificationServiceError('QUEUE_NOT_FOUND');
   }
 
+  /** Retention for terminal queue rows; pending/sending backlog is never touched. */
+  pruneFinishedQueue(olderThanMs: number): number {
+    if (!Number.isSafeInteger(olderThanMs) || olderThanMs < 0)
+      throw new RangeError('olderThanMs must be a non-negative integer');
+    const cutoff = new Date(Date.parse(this.#now()) - olderThanMs).toISOString();
+    return Number(
+      this.#db
+        .prepare(
+          "DELETE FROM notification_queue WHERE status IN ('succeeded','failed','cancelled') AND updated_at <= ?",
+        )
+        .run(cutoff).changes,
+    );
+  }
+
   deleteQueueItem(id: string): void {
     if (this.#db.prepare('DELETE FROM notification_queue WHERE id=?').run(id).changes !== 1)
       throw new CentralNotificationServiceError('QUEUE_NOT_FOUND');

apps/api/src/application/operations/delete-instance-operation.ts

@@ -1,4 +1,5 @@
 import type Database from 'better-sqlite3';
+import type { EventEnvelope } from '../events/event-journal.js';
 import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto';
 import type { Job, Preparation, PrepareOperationRequest } from '@multi-simadmin/contracts';
 import { InstanceService, InstanceServiceError } from '../instances/instance-service.js';
@@ -75,6 +76,8 @@ interface Options {
   readonly now?: () => Date;
   readonly idFactory?: () => string;
   readonly tokenFactory?: () => string;
+  /** Receives job/instance envelopes so the SSE journal can invalidate fleet views. */
+  readonly emit?: (envelope: EventEnvelope) => void;
 }
 
 const digest = (value: string): string => createHash('sha256').update(value, 'utf8').digest('hex');
@@ -93,6 +96,7 @@ export class DeleteInstanceOperation {
   private readonly clock: () => Date;
   private readonly id: () => string;
   private readonly token: () => string;
+  private readonly emit: Options['emit'];
 
   constructor(options: Options) {
     this.db = options.db;
@@ -101,6 +105,7 @@ export class DeleteInstanceOperation {
     this.clock = options.now ?? (() => new Date());
     this.id = options.idFactory ?? randomUUID;
     this.token = options.tokenFactory ?? (() => randomBytes(32).toString('base64url'));
+    this.emit = options.emit;
   }
 
   async prepare(input: PrepareOperationRequest, requestId = this.id()): Promise<Preparation> {
@@ -318,10 +323,19 @@ export class DeleteInstanceOperation {
     const release = await this.releaseBindingPort(input.instanceId, input.requestId);
     if (release?.status === 'failed') {
       this.closeAsFailed(ids, input.requestId, BINDING_REFUSAL_CODE);
+      this.#emitJob(ids.job, input.requestId);
       return this.job(ids.job);
     }
     try {
       await this.instances.delete(input.instanceId, input.revision, { allowJobHistory: true });
+      this.#emit({
+        kind: 'instance',
+        id: this.id(),
+        occurredAt: this.clock().toISOString(),
+        requestId: input.requestId,
+        instanceId: input.instanceId,
+      });
+      this.#emitJob(ids.job, input.requestId);
       const finished = this.clock().toISOString();
       this.db.transaction(() => {
         this.db
@@ -379,6 +393,25 @@ export class DeleteInstanceOperation {
   }
 
   /** Closes the job without touching the instance row, so the device stays listed and retryable. */
+  #emit(envelope: EventEnvelope): void {
+    if (!this.emit) return;
+    try {
+      this.emit(envelope);
+    } catch {
+      // A journal failure must never fail the operation itself.
+    }
+  }
+
+  #emitJob(jobId: string, requestId: string): void {
+    this.#emit({
+      kind: 'job',
+      id: this.id(),
+      occurredAt: this.clock().toISOString(),
+      requestId,
+      jobId,
+    });
+  }
+
   private closeAsFailed(
     ids: { job: string; item: string; attempt: string },
     requestId: string,

apps/api/src/application/operations/secure-operation-execution.ts

@@ -449,6 +449,24 @@ export class SecureOperationExecution {
     return this.job(ids.job);
   }
 
+  /**
+   * Retention for consumed/expired/invalidated preparations. Every prepare
+   * (manual or a scheduled retry attempt) inserts a row, so without this the
+   * table grows one row per attempt forever.
+   */
+  pruneTerminalPreparations(olderThanMs: number): number {
+    if (!Number.isSafeInteger(olderThanMs) || olderThanMs < 0)
+      throw new RangeError('olderThanMs must be a non-negative integer');
+    const cutoff = new Date(this.clock().getTime() - olderThanMs).toISOString();
+    return Number(
+      this.options.db
+        .prepare(
+          "DELETE FROM operation_preparations WHERE status IN ('consumed','expired','invalidated') AND updated_at <= ?",
+        )
+        .run(cutoff).changes,
+    );
+  }
+
   reconcileInterruptedJobs(): number {
     const now = this.clock().toISOString();
     return this.options.db.transaction(() => {

apps/api/src/application/resources/instance-resource-service.test.ts

@@ -1,4 +1,6 @@
+import Database from 'better-sqlite3';
 import { describe, expect, it } from 'vitest';
+import { migrateDatabase } from '../../infrastructure/database/migrations.js';
 
 import {
   InstanceResourceService,
@@ -155,6 +157,52 @@ describe('instance resource allowlist parsing', () => {
     ).toEqual({ simPresent: true, phoneNumbers: ['+8613800000000'] });
   });
 
+  describe('persisted resource snapshots', () => {
+    it('serves the first overview after a restart from status_snapshots with zero upstream calls', async () => {
+      const db = new Database(':memory:');
+      migrateDatabase(db);
+      // status_snapshots carries an FK to instances; seed the node it belongs to.
+      db.prepare(
+        "INSERT INTO instances (id,name,base_url,created_at,updated_at) VALUES ('device-1','device-1','http://192.168.3.55:3000','2026-01-01T00:00:00.000Z','2026-01-01T00:00:00.000Z')",
+      ).run();
+      let clock = 1_000;
+      let livePasses = 0;
+      const build = () => {
+        const requests: Array<{ url: string }> = [];
+        const service = new InstanceResourceService({
+          instances: { get: async () => ({ origin: 'http://192.168.3.55:3000' }) } as never,
+          sessions: { sessionFor: () => undefined } as never,
+          request: async (request: { url: string }) => {
+            requests.push(request);
+            return request.url.endsWith('/api/device')
+              ? response({ data: { online: true } })
+              : { status: 503, headers: {}, body: '' };
+          },
+          db,
+          now: () => clock,
+        } as never);
+        return { service, requests };
+      };
+
+      const first = build();
+      await first.service.get('device-1');
+      livePasses = first.requests.length;
+      expect(livePasses).toBe(6);
+
+      // A brand-new service (process restart) reads the persisted snapshot.
+      const second = build();
+      const resources = await second.service.get('device-1');
+      expect(second.requests).toHaveLength(0);
+      expect(resources).toEqual({ controlOnline: true });
+
+      // Past the freshness window the probe goes live again and repersists.
+      clock += 121_000;
+      await second.service.get('device-1');
+      expect(second.requests.length).toBe(livePasses);
+      db.close();
+    });
+  });
+
   describe('overview cache', () => {
     const minimal = (cacheTtlMs?: number) => {
       const requests: Array<{ url: string }> = [];

apps/api/src/application/resources/instance-resource-service.ts

@@ -1,4 +1,6 @@
 import { lookupOperator } from '@multi-simadmin/contracts';
+import { randomUUID } from 'node:crypto';
+import type Database from 'better-sqlite3';
 
 import type { InstanceService } from '../instances/instance-service.js';
 import type {
@@ -164,6 +166,8 @@ export function parseSignalStrength(response: UpstreamResponse): InstanceResourc
 }
 
 const DEFAULT_CACHE_TTL_MS = 30_000;
+const DEFAULT_SNAPSHOT_FRESH_MS = 120_000;
+const RESOURCE_SNAPSHOT_CATEGORY = 'resources';
 
 export class InstanceResourceService {
   readonly #cache = new Map<
@@ -184,6 +188,14 @@ export class InstanceResourceService {
       ) => Promise<void>;
       /** Fleet overview fan-out reads through this cache; 0 disables it. */
       readonly cacheTtlMs?: number;
+      /**
+       * When set, every live fetch is persisted into status_snapshots
+       * (category 'resources') and a snapshot younger than snapshotFreshMs is
+       * served before any upstream call, so the first overview after a restart
+       * costs zero device requests.
+       */
+      readonly db?: Database.Database;
+      readonly snapshotFreshMs?: number;
       readonly now?: () => number;
     },
   ) {}
@@ -203,11 +215,18 @@ export class InstanceResourceService {
     if (ttl > 0 && !force) {
       const cached = this.#cache.get(instanceId);
       if (cached && now - cached.fetchedAt < ttl) return cached.resources;
+      const persisted = this.#readSnapshot(instanceId, now);
+      if (persisted) {
+        this.#cache.set(instanceId, { resources: persisted, fetchedAt: now });
+        return persisted;
+      }
       const inflight = this.#inflight.get(instanceId);
       if (inflight) return inflight;
       const pending = this.#fetch(instanceId)
         .then((resources) => {
-          this.#cache.set(instanceId, { resources, fetchedAt: this.options.now?.() ?? now });
+          const fetchedAt = this.options.now?.() ?? now;
+          this.#cache.set(instanceId, { resources, fetchedAt });
+          this.#writeSnapshot(instanceId, resources, fetchedAt);
           return resources;
         })
         .finally(() => {
@@ -217,10 +236,64 @@ export class InstanceResourceService {
       return pending;
     }
     const resources = await this.#fetch(instanceId);
-    if (ttl > 0) this.#cache.set(instanceId, { resources, fetchedAt: now });
+    if (ttl > 0) {
+      this.#cache.set(instanceId, { resources, fetchedAt: now });
+      this.#writeSnapshot(instanceId, resources, now);
+    }
     return resources;
   }
 
+  #readSnapshot(instanceId: string, now: number): InstanceResources | undefined {
+    const db = this.options.db;
+    if (!db) return undefined;
+    const freshMs = this.options.snapshotFreshMs ?? DEFAULT_SNAPSHOT_FRESH_MS;
+    try {
+      const row = db
+        .prepare(
+          'SELECT payload_json, observed_at FROM status_snapshots WHERE instance_id=? AND category=?',
+        )
+        .get(instanceId, RESOURCE_SNAPSHOT_CATEGORY) as
+        | { payload_json: string; observed_at: string }
+        | undefined;
+      if (!row) return undefined;
+      const age = now - Date.parse(row.observed_at);
+      if (!Number.isFinite(age) || age < 0 || age >= freshMs) return undefined;
+      const payload: unknown = JSON.parse(row.payload_json);
+      if (typeof payload !== 'object' || payload === null || Array.isArray(payload))
+        return undefined;
+      return payload as InstanceResources;
+    } catch {
+      return undefined;
+    }
+  }
+
+  #writeSnapshot(instanceId: string, resources: InstanceResources, observedAt: number): void {
+    const db = this.options.db;
+    if (!db) return;
+    const freshMs = this.options.snapshotFreshMs ?? DEFAULT_SNAPSHOT_FRESH_MS;
+    const observed = new Date(observedAt).toISOString();
+    const expiresAt = new Date(observedAt + freshMs).toISOString();
+    try {
+      db.prepare(
+        `INSERT INTO status_snapshots (id, instance_id, category, state, payload_json, observed_at, expires_at, created_at)
+         VALUES (?, ?, ?, 'fresh', ?, ?, ?, ?)
+         ON CONFLICT(instance_id, category) DO UPDATE SET
+           id=excluded.id, state=excluded.state, payload_json=excluded.payload_json,
+           observed_at=excluded.observed_at, expires_at=excluded.expires_at, created_at=excluded.created_at`,
+      ).run(
+        randomUUID(),
+        instanceId,
+        RESOURCE_SNAPSHOT_CATEGORY,
+        JSON.stringify(resources),
+        observed,
+        expiresAt,
+        observed,
+      );
+    } catch {
+      // Persistence is an optimization; a failed write must not fail the read.
+    }
+  }
+
   async #fetch(instanceId: string): Promise<InstanceResources> {
     const [instance, session] = await Promise.all([
       this.options.instances.get(instanceId),

apps/api/src/control-plane.ts

@@ -167,6 +167,12 @@ function createHubDeviceReader(
   };
 }
 
+// Retention windows for operator-facing history. Preparations are short-lived
+// protocol state and cycle faster; audit records are the longest-lived.
+const RETENTION_HISTORY_MS = 30 * 24 * 60 * 60 * 1000;
+const RETENTION_AUDIT_MS = 90 * 24 * 60 * 60 * 1000;
+const RETENTION_PREPARATION_MS = 7 * 24 * 60 * 60 * 1000;
+
 export interface ControlPlaneApp extends FastifyInstance {
   retryPendingSecretCleanup(): Promise<unknown>;
 }
@@ -245,6 +251,7 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
     sessions,
     request: options.upstream.request,
     ensureSession,
+    db: options.db,
   });
   const messages = new InstanceMessageService({
     instances,
@@ -347,8 +354,14 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
         instances,
         releaseBinding,
         now: options.now,
+        emit: (envelope) => eventJournal.append(envelope),
       })
-    : new DeleteInstanceOperation({ db: options.db, instances, releaseBinding });
+    : new DeleteInstanceOperation({
+        db: options.db,
+        instances,
+        releaseBinding,
+        emit: (envelope) => eventJournal.append(envelope),
+      });
   deletion.reconcileInterruptedJobs();
   const listHubDevices = createHubDeviceReader(options.db, instances);
   const resolveOperationCookie = async (
@@ -511,6 +524,7 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
         resources,
         messages,
         registerDeletionPreparationRoute: false,
+        onEvent: (envelope) => eventJournal.append(envelope),
       });
       registerFleetRoutes(app, {
         instances,
@@ -585,11 +599,29 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
             .catch(() => undefined);
           // Delivered and abandoned outbox rows otherwise accumulate forever.
           void Promise.resolve()
-            .then(() => smsOutbox.prune(30 * 24 * 60 * 60 * 1000))
+            .then(() => smsOutbox.prune(RETENTION_HISTORY_MS))
             .catch(() => undefined);
           void Promise.resolve()
             .then(() => eventJournal.prune())
             .catch(() => undefined);
+          // Connection probes write a log row per beat per node; without a sweep
+          // that alone is thousands of rows a day.
+          void Promise.resolve()
+            .then(() =>
+              connectionLogs.prune({
+                before: new Date(Date.now() - RETENTION_HISTORY_MS).toISOString(),
+              }),
+            )
+            .catch(() => undefined);
+          void Promise.resolve()
+            .then(() => audit.pruneBefore(new Date(Date.now() - RETENTION_AUDIT_MS).toISOString()))
+            .catch(() => undefined);
+          void Promise.resolve()
+            .then(() => secureExecution.pruneTerminalPreparations(RETENTION_PREPARATION_MS))
+            .catch(() => undefined);
+          void Promise.resolve()
+            .then(() => centralNotifications.pruneFinishedQueue(RETENTION_HISTORY_MS))
+            .catch(() => undefined);
         }, logPruneIntervalMs)
       : undefined;
   // Scheduled component backups: the tick is cheap and only writes when a period is overdue.

apps/api/src/interface/http/instance-routes.test.ts

@@ -1,5 +1,6 @@
 import Database from 'better-sqlite3';
 import { afterEach, describe, expect, it, vi } from 'vitest';
+import type { EventEnvelope } from '../../application/events/event-journal.js';
 import { buildApp } from '../../app.js';
 import {
   InstanceSessionStore,
@@ -78,13 +79,36 @@ const fixture = () => {
     idFactory: () => `operation-${++id}`,
     tokenFactory: () => Buffer.alloc(32, 11).toString('base64url'),
   });
+  const events: Array<Record<string, unknown>> = [];
+  const onEvent = (envelope: EventEnvelope): void => {
+    events.push(envelope as unknown as Record<string, unknown>);
+  };
   const app = buildApp({
     registerRoutes: (scope) =>
-      registerInstanceRoutes(scope, { instances, connections, login, deletion }),
+      registerInstanceRoutes(scope, { instances, connections, login, deletion, onEvent }),
   });
-  return { app, instances };
+  return { app, instances, events };
 };
 
+describe('instance route SSE events', () => {
+  it('emits an instance envelope when a node is created', async () => {
+    const { app, events } = fixture();
+    const created = await app.inject({
+      method: 'POST',
+      url: '/api/v1/instances',
+      payload: {
+        name: 'Alpha',
+        origin: 'http://192.168.1.10:8080',
+        password: { action: 'set', password: 'never-return-this' },
+      },
+    });
+    expect(created.statusCode).toBe(201);
+    expect(events).toHaveLength(1);
+    expect(events[0]).toMatchObject({ kind: 'instance', requestId: expect.any(String) });
+    expect(String(events[0]?.instanceId)).toBeTruthy();
+  });
+});
+
 describe('instance HTTP routes', () => {
   it('creates, lists, gets, updates and deletes redacted instance resources', async () => {
     const { app } = fixture();

apps/api/src/interface/http/instance-routes.ts

@@ -20,6 +20,8 @@ import {
   DeleteInstanceOperationError,
 } from '../../application/operations/delete-instance-operation.js';
 import type { InstanceResourceService } from '../../application/resources/instance-resource-service.js';
+import { randomUUID } from 'node:crypto';
+import type { EventEnvelope } from '../../application/events/event-journal.js';
 import {
   MessageServiceError,
   type InstanceMessageService,
@@ -33,6 +35,8 @@ export interface InstanceRoutesOptions {
   readonly resources?: InstanceResourceService;
   readonly messages?: InstanceMessageService;
   readonly registerDeletionPreparationRoute?: boolean;
+  /** Receives instance lifecycle envelopes so the SSE journal can invalidate fleet views. */
+  readonly onEvent?: (envelope: EventEnvelope) => void;
 }
 const problem = (
   request: FastifyRequest,
@@ -56,6 +60,25 @@ const problem = (
   detail,
   requestId: request.id,
 });
+const emitInstanceEvent = (
+  options: { readonly onEvent?: (envelope: EventEnvelope) => void },
+  instanceId: string,
+  requestId: string,
+): void => {
+  if (!options.onEvent) return;
+  try {
+    options.onEvent({
+      kind: 'instance',
+      id: randomUUID(),
+      occurredAt: new Date().toISOString(),
+      requestId,
+      instanceId,
+    });
+  } catch {
+    // A journal failure must never fail the mutation itself.
+  }
+};
+
 const domainProblem = (request: FastifyRequest, error: InstanceServiceError) => {
   const status = instanceErrorStatus[error.code];
   return problem(
@@ -344,6 +367,7 @@ export function registerInstanceRoutes(app: FastifyInstance, options: InstanceRo
     async (request, reply) => {
       try {
         const created = await options.instances.create(bodyInput(request.body));
+        emitInstanceEvent(options, created.id, request.id);
         return reply.header('ETag', etag(created)).code(201).send(created);
       } catch (error) {
         if (error instanceof InstanceServiceError)
@@ -434,6 +458,7 @@ export function registerInstanceRoutes(app: FastifyInstance, options: InstanceRo
         revision(request),
         bodyPatch(request.body),
       );
+      emitInstanceEvent(options, (request.params as { instanceId: string }).instanceId, request.id);
       reply.header('ETag', etag(result));
       return result;
     }),