chick
f9186bd851
Add central notification channels, rules, queue and delivery logs, fleet organization groups and tags, device discovery, the device action catalog, instance module reads, the log centre, connection settings and system maintenance as native /api/v1 routes backed by the existing secret store, audit trail and pinned upstream transport.
470 lines
14 KiB
TypeScript
470 lines
14 KiB
TypeScript
import type Database from 'better-sqlite3';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { UpstreamError } from '../../infrastructure/transport/upstream-error.js';
|
|
|
|
export interface HealthTransportResponse {
|
|
readonly status: number;
|
|
readonly body: string;
|
|
}
|
|
|
|
export type HealthSnapshotErrorCode =
|
|
| 'AUTH_REQUIRED'
|
|
| 'UNSUPPORTED'
|
|
| 'UPSTREAM_DEGRADED'
|
|
| 'UPSTREAM_TIMEOUT'
|
|
| 'UPSTREAM_UNAVAILABLE'
|
|
| 'RESPONSE_TOO_LARGE'
|
|
| 'INVALID_RESPONSE';
|
|
|
|
export interface HealthSnapshotInstance {
|
|
readonly id: string;
|
|
readonly origin: string;
|
|
readonly revision: number;
|
|
}
|
|
|
|
export interface HealthEnvelope {
|
|
readonly status?: string;
|
|
readonly version?: string;
|
|
readonly platform?: string;
|
|
}
|
|
|
|
export type HealthSnapshotState = 'fresh' | 'stale' | 'unknown';
|
|
|
|
export interface HealthSnapshot {
|
|
readonly instanceId: string;
|
|
readonly category: 'health';
|
|
readonly state: HealthSnapshotState;
|
|
readonly payload: HealthEnvelope;
|
|
readonly observedAt: string;
|
|
readonly expiresAt: string | null;
|
|
}
|
|
|
|
export interface StatusSnapshotServiceOptions {
|
|
readonly db: Database.Database;
|
|
readonly instances: { get(instanceId: string): Promise<HealthSnapshotInstance | undefined> };
|
|
readonly transport: { get(url: string): Promise<HealthTransportResponse> };
|
|
readonly ttlMs: number;
|
|
readonly maxStaleMs: number;
|
|
readonly now?: () => Date;
|
|
readonly idFactory?: () => string;
|
|
/** Observes every probe outcome, including ones fenced from persistence by a revision change. */
|
|
readonly onProbe?: (result: HealthProbeResult) => void;
|
|
}
|
|
|
|
export interface HealthProbeResult {
|
|
readonly instanceId: string;
|
|
readonly outcome: 'success' | 'stale' | 'failed' | 'unsupported';
|
|
readonly state: HealthSnapshotState;
|
|
readonly errorCode: string | null;
|
|
readonly httpStatus: number | null;
|
|
readonly durationMs: number;
|
|
readonly observedAt: string;
|
|
}
|
|
|
|
export class StatusSnapshotError extends Error {
|
|
constructor(readonly code: 'INSTANCE_NOT_FOUND') {
|
|
super(code);
|
|
this.name = 'StatusSnapshotError';
|
|
}
|
|
}
|
|
|
|
interface SnapshotRow {
|
|
payload_json: string;
|
|
observed_at: string;
|
|
}
|
|
|
|
type SnapshotErrorCode = HealthSnapshotErrorCode;
|
|
|
|
interface PersistedHealthEnvelope {
|
|
readonly schemaVersion: 1;
|
|
readonly data: HealthEnvelope;
|
|
readonly errorCode: SnapshotErrorCode | null;
|
|
readonly httpStatus: number | null;
|
|
readonly fetchedAt: string;
|
|
readonly durationMs: number;
|
|
readonly freshness: HealthSnapshotState;
|
|
readonly supported: boolean | null;
|
|
readonly dataFetchedAt: string | null;
|
|
}
|
|
|
|
interface ClassifiedSnapshot {
|
|
readonly snapshot: HealthSnapshot;
|
|
readonly persisted: PersistedHealthEnvelope;
|
|
}
|
|
|
|
const MAX_FIELD_LENGTH = 128;
|
|
const SAFE_FIELD = /^[^\x00-\x1f\x7f]*$/;
|
|
|
|
function parseEnvelope(body: string): HealthEnvelope | undefined {
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(body);
|
|
} catch {
|
|
return undefined;
|
|
}
|
|
if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) return undefined;
|
|
const source = parsed as Record<string, unknown>;
|
|
const envelope: { status?: string; version?: string; platform?: string } = {};
|
|
for (const key of ['status', 'version', 'platform'] as const) {
|
|
const value = source[key];
|
|
if (value === undefined) continue;
|
|
if (
|
|
typeof value !== 'string' ||
|
|
value.length === 0 ||
|
|
value.length > MAX_FIELD_LENGTH ||
|
|
!SAFE_FIELD.test(value)
|
|
)
|
|
return undefined;
|
|
envelope[key] = value;
|
|
}
|
|
return Object.freeze(envelope);
|
|
}
|
|
|
|
function safePreviousPayload(payloadJson: string): HealthEnvelope | undefined {
|
|
let candidate = payloadJson;
|
|
try {
|
|
const parsed = JSON.parse(payloadJson) as { data?: unknown };
|
|
if (parsed && typeof parsed === 'object' && 'data' in parsed)
|
|
candidate = JSON.stringify(parsed.data);
|
|
} catch {
|
|
// parseEnvelope below handles malformed legacy payloads.
|
|
}
|
|
const envelope = parseEnvelope(candidate);
|
|
return envelope && Object.keys(envelope).length > 0 ? envelope : undefined;
|
|
}
|
|
|
|
function previousDataFetchedAt(previous: SnapshotRow): string {
|
|
try {
|
|
const parsed = JSON.parse(previous.payload_json) as { dataFetchedAt?: unknown };
|
|
if (
|
|
typeof parsed.dataFetchedAt === 'string' &&
|
|
Number.isFinite(Date.parse(parsed.dataFetchedAt))
|
|
) {
|
|
return parsed.dataFetchedAt;
|
|
}
|
|
} catch {
|
|
// Legacy rows use observed_at as their data fetch time.
|
|
}
|
|
return previous.observed_at;
|
|
}
|
|
|
|
function responseErrorCode(response: HealthTransportResponse | undefined): SnapshotErrorCode {
|
|
if (
|
|
!response ||
|
|
!Number.isSafeInteger(response.status) ||
|
|
response.status < 100 ||
|
|
response.status > 599
|
|
) {
|
|
return 'INVALID_RESPONSE';
|
|
}
|
|
if (response.status === 401 || response.status === 403) return 'AUTH_REQUIRED';
|
|
if (response.status === 404) return 'UNSUPPORTED';
|
|
if (response.status >= 500) return 'UPSTREAM_DEGRADED';
|
|
return 'INVALID_RESPONSE';
|
|
}
|
|
|
|
function transportErrorCode(error: unknown): SnapshotErrorCode {
|
|
if (!(error instanceof UpstreamError)) return 'UPSTREAM_UNAVAILABLE';
|
|
if (error.code === 'UPSTREAM_TIMEOUT') return 'UPSTREAM_TIMEOUT';
|
|
if (error.code === 'UPSTREAM_RESPONSE_TOO_LARGE') return 'RESPONSE_TOO_LARGE';
|
|
return 'UPSTREAM_UNAVAILABLE';
|
|
}
|
|
|
|
function isValidOwner(
|
|
instanceId: string,
|
|
owner: HealthSnapshotInstance | undefined,
|
|
): owner is HealthSnapshotInstance {
|
|
if (
|
|
!owner ||
|
|
owner.id !== instanceId ||
|
|
!Number.isSafeInteger(owner.revision) ||
|
|
owner.revision <= 0
|
|
) {
|
|
return false;
|
|
}
|
|
try {
|
|
const parsed = new URL(owner.origin);
|
|
return (
|
|
(parsed.protocol === 'http:' || parsed.protocol === 'https:') &&
|
|
parsed.origin === owner.origin &&
|
|
parsed.username === '' &&
|
|
parsed.password === ''
|
|
);
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
const healthUrl = (origin: string): string => `${origin}/api/health`;
|
|
|
|
export class StatusSnapshotService {
|
|
readonly #db: Database.Database;
|
|
readonly #instances: StatusSnapshotServiceOptions['instances'];
|
|
readonly #transport: StatusSnapshotServiceOptions['transport'];
|
|
readonly #ttlMs: number;
|
|
readonly #maxStaleMs: number;
|
|
readonly #now: () => Date;
|
|
readonly #id: () => string;
|
|
readonly #onProbe: ((result: HealthProbeResult) => void) | undefined;
|
|
readonly #latestGeneration = new Map<string, number>();
|
|
#nextGeneration = 0;
|
|
|
|
constructor(options: StatusSnapshotServiceOptions) {
|
|
if (!Number.isFinite(options.ttlMs) || options.ttlMs <= 0) {
|
|
throw new RangeError('ttlMs must be greater than zero');
|
|
}
|
|
if (!Number.isFinite(options.maxStaleMs) || options.maxStaleMs < options.ttlMs) {
|
|
throw new RangeError('maxStaleMs must be at least ttlMs');
|
|
}
|
|
this.#db = options.db;
|
|
this.#instances = options.instances;
|
|
this.#transport = options.transport;
|
|
this.#ttlMs = options.ttlMs;
|
|
this.#maxStaleMs = options.maxStaleMs;
|
|
this.#now = options.now ?? (() => new Date());
|
|
this.#id = options.idFactory ?? randomUUID;
|
|
this.#onProbe = options.onProbe;
|
|
}
|
|
|
|
async refreshHealth(instanceId: string): Promise<HealthSnapshot> {
|
|
const instance = await this.#instances.get(instanceId);
|
|
if (!isValidOwner(instanceId, instance)) throw new StatusSnapshotError('INSTANCE_NOT_FOUND');
|
|
|
|
const generation = ++this.#nextGeneration;
|
|
this.#latestGeneration.set(instanceId, generation);
|
|
|
|
const started = this.#now();
|
|
let response: HealthTransportResponse | undefined;
|
|
let transportError: unknown;
|
|
try {
|
|
response = await this.#transport.get(healthUrl(instance.origin));
|
|
} catch (error) {
|
|
transportError = error;
|
|
}
|
|
let currentOwner: HealthSnapshotInstance | undefined;
|
|
try {
|
|
currentOwner = await this.#instances.get(instanceId);
|
|
} catch {
|
|
// Owner lookup failures fence persistence, but do not discard the computed probe result.
|
|
}
|
|
const completed = this.#now();
|
|
const observedAt = completed.toISOString();
|
|
const previous = this.#previous(instanceId);
|
|
const classified = this.#classify(
|
|
instanceId,
|
|
response,
|
|
transportError,
|
|
previous,
|
|
completed,
|
|
observedAt,
|
|
Math.max(0, Math.min(Number.MAX_SAFE_INTEGER, completed.getTime() - started.getTime())),
|
|
);
|
|
const { snapshot } = classified;
|
|
|
|
this.#observe(instanceId, snapshot, classified.persisted);
|
|
if (this.#latestGeneration.get(instanceId) === generation) {
|
|
if (
|
|
isValidOwner(instanceId, currentOwner) &&
|
|
currentOwner.origin === instance.origin &&
|
|
currentOwner.revision === instance.revision
|
|
) {
|
|
this.#persist(instance, snapshot, classified.persisted);
|
|
}
|
|
this.#latestGeneration.delete(instanceId);
|
|
}
|
|
return Object.freeze({ ...snapshot, payload: Object.freeze({ ...snapshot.payload }) });
|
|
}
|
|
|
|
/** A failed connection log write must never take a health probe down with it. */
|
|
#observe(instanceId: string, snapshot: HealthSnapshot, persisted: PersistedHealthEnvelope): void {
|
|
if (!this.#onProbe) return;
|
|
const status = persisted.httpStatus;
|
|
const outcome: HealthProbeResult['outcome'] =
|
|
persisted.errorCode === 'UNSUPPORTED'
|
|
? 'unsupported'
|
|
: snapshot.state === 'fresh'
|
|
? 'success'
|
|
: snapshot.state === 'stale'
|
|
? 'stale'
|
|
: status !== null && status >= 200 && status < 300
|
|
? 'success'
|
|
: 'failed';
|
|
try {
|
|
this.#onProbe({
|
|
instanceId,
|
|
outcome,
|
|
state: snapshot.state,
|
|
errorCode: persisted.errorCode,
|
|
httpStatus: status,
|
|
durationMs: persisted.durationMs,
|
|
observedAt: snapshot.observedAt,
|
|
});
|
|
} catch {
|
|
// Diagnostics are advisory; the snapshot result stands on its own.
|
|
}
|
|
}
|
|
|
|
#previous(instanceId: string): SnapshotRow | undefined {
|
|
return this.#db
|
|
.prepare(
|
|
"SELECT payload_json,observed_at FROM status_snapshots WHERE instance_id=? AND category='health'",
|
|
)
|
|
.get(instanceId) as SnapshotRow | undefined;
|
|
}
|
|
|
|
#classify(
|
|
instanceId: string,
|
|
response: HealthTransportResponse | undefined,
|
|
transportError: unknown,
|
|
previous: SnapshotRow | undefined,
|
|
completed: Date,
|
|
observedAt: string,
|
|
durationMs: number,
|
|
): ClassifiedSnapshot {
|
|
if (
|
|
response &&
|
|
Number.isSafeInteger(response.status) &&
|
|
response.status >= 200 &&
|
|
response.status < 300 &&
|
|
typeof response.body === 'string'
|
|
) {
|
|
const payload = parseEnvelope(response.body);
|
|
if (payload) {
|
|
const snapshot: HealthSnapshot = {
|
|
instanceId,
|
|
category: 'health',
|
|
state: 'fresh',
|
|
payload,
|
|
observedAt,
|
|
expiresAt: new Date(completed.getTime() + this.#ttlMs).toISOString(),
|
|
};
|
|
return {
|
|
snapshot,
|
|
persisted: this.#envelope(snapshot, null, response.status, durationMs, true, observedAt),
|
|
};
|
|
}
|
|
}
|
|
|
|
if (response?.status === 404) {
|
|
const snapshot = this.#unknown(instanceId, observedAt);
|
|
return {
|
|
snapshot,
|
|
persisted: this.#envelope(snapshot, 'UNSUPPORTED', 404, durationMs, false, null),
|
|
};
|
|
}
|
|
|
|
const errorCode: SnapshotErrorCode = transportError
|
|
? transportErrorCode(transportError)
|
|
: responseErrorCode(response);
|
|
const httpStatus =
|
|
response &&
|
|
Number.isSafeInteger(response.status) &&
|
|
response.status >= 100 &&
|
|
response.status <= 599
|
|
? response.status
|
|
: null;
|
|
const previousPayload = previous && safePreviousPayload(previous.payload_json);
|
|
const previousTime = previous ? Date.parse(previousDataFetchedAt(previous)) : Number.NaN;
|
|
if (
|
|
previousPayload &&
|
|
Number.isFinite(previousTime) &&
|
|
completed.getTime() - previousTime <= this.#maxStaleMs
|
|
) {
|
|
const snapshot: HealthSnapshot = {
|
|
instanceId,
|
|
category: 'health',
|
|
state: 'stale',
|
|
payload: previousPayload,
|
|
observedAt,
|
|
expiresAt: new Date(previousTime + this.#maxStaleMs).toISOString(),
|
|
};
|
|
return {
|
|
snapshot,
|
|
persisted: this.#envelope(
|
|
snapshot,
|
|
errorCode,
|
|
httpStatus,
|
|
durationMs,
|
|
httpStatus === null ? null : true,
|
|
previousDataFetchedAt(previous),
|
|
),
|
|
};
|
|
}
|
|
const snapshot = this.#unknown(instanceId, observedAt);
|
|
return {
|
|
snapshot,
|
|
persisted: this.#envelope(
|
|
snapshot,
|
|
errorCode,
|
|
httpStatus,
|
|
durationMs,
|
|
httpStatus === null ? null : true,
|
|
null,
|
|
),
|
|
};
|
|
}
|
|
|
|
#unknown(instanceId: string, observedAt: string): HealthSnapshot {
|
|
return {
|
|
instanceId,
|
|
category: 'health',
|
|
state: 'unknown',
|
|
payload: {},
|
|
observedAt,
|
|
expiresAt: null,
|
|
};
|
|
}
|
|
|
|
#envelope(
|
|
snapshot: HealthSnapshot,
|
|
errorCode: SnapshotErrorCode | null,
|
|
httpStatus: number | null,
|
|
durationMs: number,
|
|
supported: boolean | null,
|
|
dataFetchedAt: string | null,
|
|
): PersistedHealthEnvelope {
|
|
return {
|
|
schemaVersion: 1,
|
|
data: snapshot.payload,
|
|
errorCode,
|
|
httpStatus,
|
|
fetchedAt: snapshot.observedAt,
|
|
durationMs,
|
|
freshness: snapshot.state,
|
|
supported,
|
|
dataFetchedAt,
|
|
};
|
|
}
|
|
|
|
#persist(
|
|
owner: HealthSnapshotInstance,
|
|
snapshot: HealthSnapshot,
|
|
persisted: PersistedHealthEnvelope,
|
|
): void {
|
|
this.#db
|
|
.prepare(
|
|
`INSERT INTO status_snapshots
|
|
(id,instance_id,category,state,payload_json,observed_at,expires_at,created_at)
|
|
SELECT ?,id,'health',?,?,?,?,? FROM instances
|
|
WHERE id=? AND base_url=? AND config_revision=?
|
|
ON CONFLICT(instance_id,category) DO UPDATE SET
|
|
state=excluded.state,
|
|
payload_json=excluded.payload_json,
|
|
observed_at=excluded.observed_at,
|
|
expires_at=excluded.expires_at
|
|
WHERE excluded.observed_at >= status_snapshots.observed_at`,
|
|
)
|
|
.run(
|
|
this.#id(),
|
|
snapshot.state,
|
|
JSON.stringify(persisted),
|
|
snapshot.observedAt,
|
|
snapshot.expiresAt,
|
|
snapshot.observedAt,
|
|
snapshot.instanceId,
|
|
owner.origin,
|
|
owner.revision,
|
|
);
|
|
}
|
|
}
|