feat: complete TASK-WP2-06 generation terminal states
This commit is contained in:
suyx
@@ -0,0 +1,241 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
import Database from "better-sqlite3";
|
||||
import type BetterSqlite3 from "better-sqlite3";
|
||||
|
||||
import { configureWorkerDatabase } from "./sqlite-connection.js";
|
||||
|
||||
const auditRetentionMilliseconds = 180 * 24 * 60 * 60 * 1_000;
|
||||
const accountRefPattern = /^[a-z][a-z0-9_-]{2,119}$/;
|
||||
const sourceCategoryPattern = /^[a-z][a-z0-9_]{0,79}$/;
|
||||
|
||||
interface RuntimeRow {
|
||||
available_for_new_jobs: number;
|
||||
balance_blocked: number;
|
||||
gateway_account_ref: string;
|
||||
model_id: string;
|
||||
runtime_reason: string;
|
||||
runtime_version: number;
|
||||
updated_at: number;
|
||||
}
|
||||
|
||||
interface BalanceRow {
|
||||
affected_model_ids_json: string;
|
||||
balance_status: "available" | "insufficient" | "unknown";
|
||||
detected_at: number;
|
||||
gateway_account_ref: string;
|
||||
impact_scope: "model" | "account" | "unknown";
|
||||
last_confirmed_at: number | null;
|
||||
recovery_status: "not_required" | "awaiting_confirmation" | "confirmed";
|
||||
runtime_unavailable_model_ids_json: string;
|
||||
source_category: string;
|
||||
}
|
||||
|
||||
export interface GatewayBalanceView {
|
||||
affectedModelIds: string[];
|
||||
balanceStatus: "available" | "insufficient" | "unknown";
|
||||
detectedAt: string;
|
||||
gatewayAccountRef: string;
|
||||
impactScope: "model" | "account" | "unknown";
|
||||
lastConfirmedAt: string | null;
|
||||
recoveryStatus: "not_required" | "awaiting_confirmation" | "confirmed";
|
||||
runtimeUnavailableModelIds: string[];
|
||||
sourceCategory: string;
|
||||
}
|
||||
|
||||
export class GatewayBalanceRuntime {
|
||||
readonly database: BetterSqlite3.Database;
|
||||
private readonly clock: () => number;
|
||||
private readonly ownsDatabase: boolean;
|
||||
|
||||
constructor(input: { clock?: () => number; database?: BetterSqlite3.Database; databasePath?: string }) {
|
||||
if (!input.database && !input.databasePath) throw new Error("gateway_balance_database_required");
|
||||
this.database = input.database ?? new Database(input.databasePath!);
|
||||
this.ownsDatabase = !input.database;
|
||||
this.clock = input.clock ?? Date.now;
|
||||
if (this.ownsDatabase) configureWorkerDatabase(this.database);
|
||||
this.migrate();
|
||||
}
|
||||
|
||||
close() {
|
||||
if (this.ownsDatabase) this.database.close();
|
||||
}
|
||||
|
||||
seedModels(models: Array<{ gatewayAccountRef: string; modelId: string }>) {
|
||||
const now = this.clock();
|
||||
this.immediate(() => {
|
||||
for (const model of models) {
|
||||
if (!accountRefPattern.test(model.gatewayAccountRef) || !model.modelId) throw new Error("gateway_balance_model_invalid");
|
||||
this.database.prepare(`
|
||||
INSERT INTO model_runtime_availability (
|
||||
model_id, gateway_account_ref, balance_blocked, available_for_new_jobs, runtime_reason, runtime_version, updated_at
|
||||
) VALUES (?, ?, 0, 1, 'available', 1, ?)
|
||||
ON CONFLICT(model_id) DO UPDATE SET gateway_account_ref = excluded.gateway_account_ref
|
||||
`).run(model.modelId, model.gatewayAccountRef, now);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
recordInsufficient(input: {
|
||||
eventId: string;
|
||||
gatewayAccountRef: string;
|
||||
impactScope: "model" | "account" | "unknown";
|
||||
modelId: string;
|
||||
sourceCategory: string;
|
||||
}) {
|
||||
if (!input.eventId || !accountRefPattern.test(input.gatewayAccountRef) || !input.modelId
|
||||
|| !sourceCategoryPattern.test(input.sourceCategory)) throw new Error("gateway_balance_signal_invalid");
|
||||
return this.immediate(() => {
|
||||
const receipt = this.database.prepare("SELECT response_json FROM gateway_balance_event_receipts WHERE event_id = ?")
|
||||
.get(input.eventId) as { response_json: string } | undefined;
|
||||
if (receipt) return JSON.parse(receipt.response_json) as GatewayBalanceView;
|
||||
this.database.prepare(`
|
||||
INSERT INTO model_runtime_availability (
|
||||
model_id, gateway_account_ref, balance_blocked, available_for_new_jobs, runtime_reason, runtime_version, updated_at
|
||||
) VALUES (?, ?, 0, 1, 'available', 1, ?)
|
||||
ON CONFLICT(model_id) DO UPDATE SET gateway_account_ref = excluded.gateway_account_ref
|
||||
`).run(input.modelId, input.gatewayAccountRef, this.clock());
|
||||
const rows = this.database.prepare("SELECT model_id FROM model_runtime_availability WHERE gateway_account_ref = ? ORDER BY model_id")
|
||||
.all(input.gatewayAccountRef) as Array<{ model_id: string }>;
|
||||
const newlyAffected = input.impactScope === "model" ? [input.modelId] : rows.map((row) => row.model_id);
|
||||
const existing = this.database.prepare("SELECT runtime_unavailable_model_ids_json FROM gateway_balance_states WHERE gateway_account_ref = ?")
|
||||
.get(input.gatewayAccountRef) as { runtime_unavailable_model_ids_json: string } | undefined;
|
||||
const unavailable = [...new Set([...(existing ? JSON.parse(existing.runtime_unavailable_model_ids_json) as string[] : []), ...newlyAffected])].toSorted();
|
||||
const now = this.clock();
|
||||
for (const modelId of unavailable) {
|
||||
this.database.prepare(`
|
||||
UPDATE model_runtime_availability
|
||||
SET balance_blocked = 1, available_for_new_jobs = 0, runtime_reason = 'gateway_balance_insufficient',
|
||||
runtime_version = runtime_version + 1, updated_at = ?
|
||||
WHERE model_id = ? AND gateway_account_ref = ? AND balance_blocked = 0
|
||||
`).run(now, modelId, input.gatewayAccountRef);
|
||||
}
|
||||
this.database.prepare(`
|
||||
INSERT INTO gateway_balance_states (
|
||||
gateway_account_ref, balance_status, impact_scope, affected_model_ids_json,
|
||||
runtime_unavailable_model_ids_json, detected_at, last_confirmed_at, recovery_status, source_category
|
||||
) VALUES (?, 'insufficient', ?, ?, ?, ?, NULL, 'awaiting_confirmation', ?)
|
||||
ON CONFLICT(gateway_account_ref) DO UPDATE SET
|
||||
balance_status = 'insufficient', impact_scope = excluded.impact_scope,
|
||||
affected_model_ids_json = excluded.affected_model_ids_json,
|
||||
runtime_unavailable_model_ids_json = excluded.runtime_unavailable_model_ids_json,
|
||||
detected_at = excluded.detected_at, last_confirmed_at = NULL,
|
||||
recovery_status = 'awaiting_confirmation', source_category = excluded.source_category
|
||||
`).run(input.gatewayAccountRef, input.impactScope, JSON.stringify(newlyAffected.toSorted()), JSON.stringify(unavailable), now, input.sourceCategory);
|
||||
this.database.prepare(`
|
||||
INSERT INTO admin_operation_logs (
|
||||
log_id, actor_type, actor_ref, operation_type, target_type, target_ref,
|
||||
result, before_summary, after_summary, occurred_at, expires_at
|
||||
) VALUES (?, 'system', 'generation_worker', 'gateway_balance_detected', 'gateway_balance_state', ?, 'succeeded', NULL, ?, ?, ?)
|
||||
`).run(randomUUID(), input.gatewayAccountRef, JSON.stringify({ impact_scope: input.impactScope, runtime_unavailable_model_ids: unavailable }), now, now + auditRetentionMilliseconds);
|
||||
const result = this.readState(input.gatewayAccountRef)!;
|
||||
this.database.prepare("INSERT INTO gateway_balance_event_receipts (event_id, gateway_account_ref, response_json, created_at) VALUES (?, ?, ?, ?)")
|
||||
.run(input.eventId, input.gatewayAccountRef, JSON.stringify(result), now);
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
readState(gatewayAccountRef: string) {
|
||||
const row = this.database.prepare("SELECT * FROM gateway_balance_states WHERE gateway_account_ref = ?").get(gatewayAccountRef) as BalanceRow | undefined;
|
||||
return row ? this.view(row) : undefined;
|
||||
}
|
||||
|
||||
listRuntime() {
|
||||
const rows = this.database.prepare("SELECT * FROM model_runtime_availability ORDER BY model_id").all() as RuntimeRow[];
|
||||
return rows.map((row) => ({
|
||||
availableForNewJobs: row.available_for_new_jobs === 1,
|
||||
balanceBlocked: row.balance_blocked === 1,
|
||||
gatewayAccountRef: row.gateway_account_ref,
|
||||
modelId: row.model_id,
|
||||
runtimeReason: row.runtime_reason,
|
||||
runtimeVersion: row.runtime_version,
|
||||
updatedAt: new Date(row.updated_at).toISOString(),
|
||||
}));
|
||||
}
|
||||
|
||||
restoreWithoutConfirmedRecovery(gatewayAccountRef: string, _actorId: string) {
|
||||
const state = this.readState(gatewayAccountRef);
|
||||
if (!state || state.recoveryStatus !== "confirmed") throw new Error("gateway_balance_recovery_unconfirmed");
|
||||
throw new Error("gateway_balance_recovery_owned_by_wp3");
|
||||
}
|
||||
|
||||
private view(row: BalanceRow): GatewayBalanceView {
|
||||
return {
|
||||
affectedModelIds: JSON.parse(row.affected_model_ids_json) as string[],
|
||||
balanceStatus: row.balance_status,
|
||||
detectedAt: new Date(row.detected_at).toISOString(),
|
||||
gatewayAccountRef: row.gateway_account_ref,
|
||||
impactScope: row.impact_scope,
|
||||
lastConfirmedAt: row.last_confirmed_at === null ? null : new Date(row.last_confirmed_at).toISOString(),
|
||||
recoveryStatus: row.recovery_status,
|
||||
runtimeUnavailableModelIds: JSON.parse(row.runtime_unavailable_model_ids_json) as string[],
|
||||
sourceCategory: row.source_category,
|
||||
};
|
||||
}
|
||||
|
||||
private immediate<T>(action: () => T) {
|
||||
if (this.database.inTransaction) return action();
|
||||
this.database.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
const result = action();
|
||||
this.database.exec("COMMIT");
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (this.database.inTransaction) this.database.exec("ROLLBACK");
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private migrate() {
|
||||
this.database.exec(`
|
||||
CREATE TABLE IF NOT EXISTS model_runtime_availability (
|
||||
model_id TEXT PRIMARY KEY,
|
||||
gateway_account_ref TEXT NOT NULL,
|
||||
balance_blocked INTEGER NOT NULL CHECK (balance_blocked IN (0, 1)),
|
||||
available_for_new_jobs INTEGER NOT NULL CHECK (available_for_new_jobs IN (0, 1)),
|
||||
runtime_reason TEXT NOT NULL CHECK (runtime_reason IN (
|
||||
'available', 'configured_disabled', 'contract_unverified', 'contract_blocked',
|
||||
'gateway_balance_insufficient', 'gateway_paused', 'worker_degraded'
|
||||
)),
|
||||
runtime_version INTEGER NOT NULL CHECK (runtime_version > 0),
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS gateway_balance_states (
|
||||
gateway_account_ref TEXT PRIMARY KEY,
|
||||
balance_status TEXT NOT NULL CHECK (balance_status IN ('available', 'insufficient', 'unknown')),
|
||||
impact_scope TEXT NOT NULL CHECK (impact_scope IN ('model', 'account', 'unknown')),
|
||||
affected_model_ids_json TEXT NOT NULL CHECK (json_valid(affected_model_ids_json)),
|
||||
runtime_unavailable_model_ids_json TEXT NOT NULL CHECK (json_valid(runtime_unavailable_model_ids_json)),
|
||||
detected_at INTEGER NOT NULL,
|
||||
last_confirmed_at INTEGER,
|
||||
recovery_status TEXT NOT NULL CHECK (recovery_status IN ('not_required', 'awaiting_confirmation', 'confirmed')),
|
||||
source_category TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS gateway_balance_event_receipts (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
gateway_account_ref TEXT NOT NULL,
|
||||
response_json TEXT NOT NULL CHECK (json_valid(response_json)),
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS admin_operation_logs (
|
||||
log_id TEXT PRIMARY KEY,
|
||||
actor_type TEXT NOT NULL CHECK (actor_type IN ('system', 'super_admin')),
|
||||
actor_ref TEXT NOT NULL CHECK (dada_audit_ref_is_safe(actor_ref) = 1),
|
||||
operation_type TEXT NOT NULL CHECK (dada_audit_ref_is_safe(operation_type) = 1),
|
||||
target_type TEXT NOT NULL CHECK (dada_audit_ref_is_safe(target_type) = 1),
|
||||
target_ref TEXT NOT NULL CHECK (dada_audit_ref_is_safe(target_ref) = 1),
|
||||
result TEXT NOT NULL CHECK (result IN ('succeeded', 'failed')),
|
||||
before_summary TEXT CHECK (before_summary IS NULL OR dada_audit_summary_is_safe(before_summary) = 1),
|
||||
after_summary TEXT CHECK (after_summary IS NULL OR dada_audit_summary_is_safe(after_summary) = 1),
|
||||
occurred_at INTEGER NOT NULL,
|
||||
expires_at INTEGER NOT NULL CHECK (expires_at = occurred_at + ${auditRetentionMilliseconds})
|
||||
);
|
||||
CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_update
|
||||
BEFORE UPDATE ON admin_operation_logs BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END;
|
||||
CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_delete
|
||||
BEFORE DELETE ON admin_operation_logs
|
||||
WHEN dada_allow_retention_purge() <> 1 OR OLD.expires_at > dada_retention_purge_now()
|
||||
BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END;
|
||||
`);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user