diff --git a/README.md b/README.md index aca658d..2d48afe 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,16 @@ sh /tmp/multi-simadmin-install.sh uninstall `uninstall` 默认只移除程序源码，保留数据库、Gateway token 和日志。默认安装位置 macOS 为 `~/Library/Application Support/multi-simadmin`，Linux 为 `~/.local/share/multi-simadmin`（遵循 XDG），均可通过 `MULTI_SIMADMIN_HOME` 覆盖；额外 LAN Host 可通过 `MULTI_SIMADMIN_ALLOWED_HOSTS`（逗号分隔）配置。 +#### Linux systemd（可选，但推荐） + +`sh install.sh unit` 会生成 `~/.config/systemd/user/` 下的 API 与 Gateway 用户单元（`Restart=on-failure`、开机自启、密钥经 0600 的 `service-env` 注入）。随后： + +```bash +systemctl --user enable --now multi-simadmin-api.service multi-simadmin-gateway.service +``` + +单元内声明了 `MULTI_SIMADMIN_SYSTEMD_UNIT`，因此 Web 控制台的“在线更新”在 Linux 上会把重启交给 systemd 完成，形成自更新闭环。由 systemd 托管后请用 `systemctl` 管理进程，不要再用本脚本 `start/stop`，避免双进程争抢端口。 + ### Fleet 节点总览与跨域健康 `/fleet` 读取本机 Control Plane 中的节点状态、资源快照与设备号码，形成节点总览。侧栏在实例总数 / 在线 / 需处理之外，还聚合跨域健康指标： diff --git a/apps/api/src/application/audit/audit-query-service.ts b/apps/api/src/application/audit/audit-query-service.ts index f799e14..f3e91c8 100644 --- a/apps/api/src/application/audit/audit-query-service.ts +++ b/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 diff --git a/apps/api/src/application/notifications/central-notification-service.ts b/apps/api/src/application/notifications/central-notification-service.ts index 6ff9d1e..704ec76 100644 --- a/apps/api/src/application/notifications/central-notification-service.ts +++ b/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'); diff --git a/apps/api/src/application/operations/delete-instance-operation.ts b/apps/api/src/application/operations/delete-instance-operation.ts index 29b70de..b04fcfe 100644 --- a/apps/api/src/application/operations/delete-instance-operation.ts +++ b/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 { @@ -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, diff --git a/apps/api/src/application/operations/secure-operation-execution.ts b/apps/api/src/application/operations/secure-operation-execution.ts index f8d9fd1..fa0915d 100644 --- a/apps/api/src/application/operations/secure-operation-execution.ts +++ b/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(() => { diff --git a/apps/api/src/application/resources/instance-resource-service.test.ts b/apps/api/src/application/resources/instance-resource-service.test.ts index 7ca5039..30bead7 100644 --- a/apps/api/src/application/resources/instance-resource-service.test.ts +++ b/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 }> = []; diff --git a/apps/api/src/application/resources/instance-resource-service.ts b/apps/api/src/application/resources/instance-resource-service.ts index 718dd96..043b157 100644 --- a/apps/api/src/application/resources/instance-resource-service.ts +++ b/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; /** 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 { const [instance, session] = await Promise.all([ this.options.instances.get(instanceId), diff --git a/apps/api/src/control-plane.ts b/apps/api/src/control-plane.ts index 3e8541c..78679cb 100644 --- a/apps/api/src/control-plane.ts +++ b/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; } @@ -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. diff --git a/apps/api/src/interface/http/instance-routes.test.ts b/apps/api/src/interface/http/instance-routes.test.ts index 8875861..10f3f78 100644 --- a/apps/api/src/interface/http/instance-routes.test.ts +++ b/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> = []; + const onEvent = (envelope: EventEnvelope): void => { + events.push(envelope as unknown as Record); + }; 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(); diff --git a/apps/api/src/interface/http/instance-routes.ts b/apps/api/src/interface/http/instance-routes.ts index a91b556..4ca7d73 100644 --- a/apps/api/src/interface/http/instance-routes.ts +++ b/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; }), diff --git a/scripts/install.sh b/scripts/install.sh index 08f2edb..e219ea2 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -36,6 +36,7 @@ Multi SimAdmin 一键安装与服务管理 restart 停止后重新启动 status 显示进程、端口与健康状态 uninstall 卸载程序，默认保留数据、令牌和日志 + unit （仅 Linux）生成 systemd 用户单元，支持开机自启与在线自更新重启 环境变量： MULTI_SIMADMIN_HOME 安装根目录（默认 macOS 为 ~/Library/Application Support/multi-simadmin，Linux 为 ~/.local/share/multi-simadmin） MULTI_SIMADMIN_ALLOWED_HOSTS 额外允许的 LAN 主机名/IP，逗号分隔 @@ -297,6 +298,72 @@ uninstall_app() { say "程序已卸载；默认保留数据、Gateway token 和日志：$APP_ROOT" } +write_systemd_units() { + require_platform + [ "$PLATFORM" = Linux ] || die "systemd 单元仅适用于 Linux" + command -v systemctl >/dev/null 2>&1 || die "缺少 systemctl" + [ -f "$SOURCE_DIR/apps/web/dist/index.html" ] || die "尚未安装，请先运行 install" + [ -f "$TOKEN_FILE" ] || die "缺少 Gateway token，请先运行 install" + node_bin=$(command -v node) + unit_dir=${XDG_CONFIG_HOME:-$HOME/.config}/systemd/user + mkdir -p "$unit_dir" + chmod 700 "$unit_dir" + + env_file="$APP_ROOT/service-env" + { printf 'MULTI_SIMADMIN_GATEWAY_TOKEN='; cat "$TOKEN_FILE"; } >"$env_file" + chmod 600 "$env_file" + + cat >"$unit_dir/multi-simadmin-api.service" <"$unit_dir/multi-simadmin-gateway.service" <&2; exit 2 ;; esac