914 lines
28 KiB
TypeScript
914 lines
28 KiB
TypeScript
import log from "electron-log";
|
|
import { getDeviceId } from "../system/deviceId";
|
|
import { getDomainTokenKey } from "@shared/utils/domain";
|
|
import { ensureDigitalEmployeeSchema, getDb, readSetting } from "../../db";
|
|
|
|
const DEFAULT_SYNC_PATH = "/api/digital-employee/sync/outbox";
|
|
const MANAGEMENT_SYNC_RECORD_PATH =
|
|
"/system/content/content-digital-employee-sync";
|
|
const DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION = "qimingclaw.digital_employee.sync.v1";
|
|
const DEFAULT_INTERVAL_MS = 30_000;
|
|
const DEFAULT_BATCH_SIZE = 50;
|
|
const REQUEST_TIMEOUT_MS = 15_000;
|
|
const MAX_RETRY_DELAY_MS = 30 * 60_000;
|
|
const FAILED_RETRY_CANDIDATE_MULTIPLIER = 4;
|
|
const ATTEMPT_HISTORY_RETENTION_MS = 30 * 24 * 60 * 60_000;
|
|
|
|
let syncTimer: ReturnType<typeof setInterval> | null = null;
|
|
let syncing = false;
|
|
let lastSyncAt: string | null = null;
|
|
let lastSyncError: string | null = null;
|
|
|
|
function getDigitalEmployeeDb(): ReturnType<typeof getDb> {
|
|
if (!ensureDigitalEmployeeSchema()) return null;
|
|
return getDb();
|
|
}
|
|
|
|
interface DigitalEmployeeSyncConfig {
|
|
enabled: boolean;
|
|
endpoint: string | null;
|
|
clientKey: string | null;
|
|
authToken: string | null;
|
|
batchSize: number;
|
|
}
|
|
|
|
interface OutboxRow {
|
|
id: string;
|
|
entity_type: string;
|
|
entity_id: string;
|
|
operation: string;
|
|
payload: string;
|
|
attempts: number;
|
|
last_attempt_at?: string | null;
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncAttempt {
|
|
attemptNo: number;
|
|
status: string;
|
|
error: string | null;
|
|
endpoint: string | null;
|
|
startedAt: string;
|
|
finishedAt: string;
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncFailureGroup {
|
|
entityType: string;
|
|
count: number;
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncFailureSummary {
|
|
total: number;
|
|
dueForRetry: number;
|
|
backoff: number;
|
|
byEntityType: DigitalEmployeeSyncFailureGroup[];
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncFailure {
|
|
id: string;
|
|
entityType: string;
|
|
entityId: string;
|
|
entitySummary: DigitalEmployeeSyncEntitySummary | null;
|
|
operation: string;
|
|
attempts: number;
|
|
lastAttemptAt: string | null;
|
|
nextRetryAt: string | null;
|
|
dueForRetry: boolean;
|
|
managementRecordUrl: string | null;
|
|
attemptHistory: DigitalEmployeeSyncAttempt[];
|
|
error: string | null;
|
|
createdAt: string;
|
|
updatedAt: string;
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncEntitySummary {
|
|
title: string | null;
|
|
status: string | null;
|
|
summary: string | null;
|
|
}
|
|
|
|
interface SyncAck {
|
|
outbox_id?: string;
|
|
id?: string;
|
|
entity_type?: string;
|
|
entity_id?: string;
|
|
remote_id?: string;
|
|
}
|
|
|
|
interface SyncResponse {
|
|
success?: boolean;
|
|
code?: string | number;
|
|
status?: string | number;
|
|
acked?: SyncAck[];
|
|
data?: {
|
|
acked?: SyncAck[];
|
|
};
|
|
message?: string;
|
|
}
|
|
|
|
export interface DigitalEmployeeSyncStatus {
|
|
running: boolean;
|
|
syncing: boolean;
|
|
enabled: boolean;
|
|
endpoint: string | null;
|
|
missingCredentials: string[];
|
|
pending: number;
|
|
failed: number;
|
|
lastSyncAt: string | null;
|
|
lastSyncError: string | null;
|
|
failureSummary: DigitalEmployeeSyncFailureSummary;
|
|
recentFailures: DigitalEmployeeSyncFailure[];
|
|
}
|
|
|
|
export function startDigitalEmployeeSyncWorker(): void {
|
|
if (syncTimer) return;
|
|
syncTimer = setInterval(() => {
|
|
void flushDigitalEmployeeSyncOutbox().catch((error) => {
|
|
lastSyncError = normalizeError(error);
|
|
log.warn("[DigitalEmployeeSync] Scheduled sync failed:", error);
|
|
});
|
|
}, DEFAULT_INTERVAL_MS);
|
|
void flushDigitalEmployeeSyncOutbox().catch((error) => {
|
|
lastSyncError = normalizeError(error);
|
|
log.debug("[DigitalEmployeeSync] Initial sync skipped/failed:", error);
|
|
});
|
|
log.info("[DigitalEmployeeSync] Worker started");
|
|
}
|
|
|
|
export function stopDigitalEmployeeSyncWorker(): void {
|
|
if (!syncTimer) return;
|
|
clearInterval(syncTimer);
|
|
syncTimer = null;
|
|
log.info("[DigitalEmployeeSync] Worker stopped");
|
|
}
|
|
|
|
export function getDigitalEmployeeSyncStatus(): DigitalEmployeeSyncStatus {
|
|
const config = resolveSyncConfig();
|
|
return {
|
|
running: Boolean(syncTimer),
|
|
syncing,
|
|
enabled: config.enabled,
|
|
endpoint: config.endpoint,
|
|
missingCredentials: missingSyncCredentialLabels(config),
|
|
pending: countOutbox("pending"),
|
|
failed: countOutbox("failed"),
|
|
lastSyncAt,
|
|
lastSyncError,
|
|
failureSummary: readFailureSummary(),
|
|
recentFailures: readRecentFailures(config),
|
|
};
|
|
}
|
|
|
|
export async function flushDigitalEmployeeSyncOutbox(
|
|
options: { force?: boolean } = {},
|
|
): Promise<DigitalEmployeeSyncStatus> {
|
|
if (syncing) return getDigitalEmployeeSyncStatus();
|
|
|
|
const config = resolveSyncConfig();
|
|
if (
|
|
!config.enabled ||
|
|
!config.endpoint ||
|
|
missingSyncCredentialLabels(config).length > 0
|
|
) {
|
|
lastSyncError = null;
|
|
return getDigitalEmployeeSyncStatus();
|
|
}
|
|
|
|
const rows = readPendingOutbox(config.batchSize, options.force === true);
|
|
if (rows.length === 0) {
|
|
lastSyncError = null;
|
|
return getDigitalEmployeeSyncStatus();
|
|
}
|
|
|
|
syncing = true;
|
|
const startedAt = new Date().toISOString();
|
|
markRowsSyncing(rows, startedAt);
|
|
|
|
try {
|
|
const response = await postOutbox(config, rows);
|
|
const finishedAt = new Date().toISOString();
|
|
const acked = response.data?.acked || response.acked || [];
|
|
const responseError = readSyncResponseError(response);
|
|
if (responseError) throw new Error(responseError);
|
|
const syncedRows = acked.length > 0
|
|
? rows.filter((row) => acked.some((ack) => ackMatchesRow(ack, row)))
|
|
: rows;
|
|
const unackedRows = acked.length > 0
|
|
? rows.filter((row) => !acked.some((ack) => ackMatchesRow(ack, row)))
|
|
: [];
|
|
recordSyncAttempts(syncedRows, config, "synced", null, startedAt, finishedAt);
|
|
if (acked.length > 0) {
|
|
markRowsSynced(acked, finishedAt, rows);
|
|
} else {
|
|
markRowsSynced(
|
|
rows.map((row) => ({
|
|
outbox_id: row.id,
|
|
entity_type: row.entity_type,
|
|
entity_id: row.entity_id,
|
|
})),
|
|
finishedAt,
|
|
);
|
|
}
|
|
if (unackedRows.length > 0) {
|
|
const unackedError = "management sync response did not acknowledge item";
|
|
recordSyncAttempts(
|
|
unackedRows,
|
|
config,
|
|
"failed",
|
|
unackedError,
|
|
startedAt,
|
|
finishedAt,
|
|
);
|
|
markRowsFailed(unackedRows, finishedAt, unackedError);
|
|
lastSyncError = unackedError;
|
|
} else {
|
|
lastSyncError = null;
|
|
}
|
|
lastSyncAt = finishedAt;
|
|
} catch (error) {
|
|
const finishedAt = new Date().toISOString();
|
|
lastSyncError = normalizeError(error);
|
|
recordSyncAttempts(rows, config, "failed", lastSyncError, startedAt, finishedAt);
|
|
markRowsFailed(rows, finishedAt, lastSyncError);
|
|
log.warn("[DigitalEmployeeSync] Flush failed:", lastSyncError);
|
|
} finally {
|
|
syncing = false;
|
|
}
|
|
|
|
return getDigitalEmployeeSyncStatus();
|
|
}
|
|
|
|
function readSyncResponseError(response: SyncResponse): string | null {
|
|
if (response.success === false) {
|
|
return response.message || "digital employee sync rejected";
|
|
}
|
|
|
|
const code = normalizeResponseCode(response.code);
|
|
if (code && code !== "0000") {
|
|
return response.message || `digital employee sync rejected: ${code}`;
|
|
}
|
|
|
|
const status = normalizeResponseCode(response.status);
|
|
if (status && ["error", "fail", "failed", "failure"].includes(status.toLowerCase())) {
|
|
return response.message || "digital employee sync rejected";
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function normalizeResponseCode(value: unknown): string | null {
|
|
if (typeof value === "string") {
|
|
const trimmed = value.trim();
|
|
return trimmed || null;
|
|
}
|
|
if (typeof value === "number" && Number.isFinite(value)) {
|
|
return String(value);
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function resolveSyncConfig(): DigitalEmployeeSyncConfig {
|
|
const config = (readSetting("digital_employee_sync_config") || {}) as Record<
|
|
string,
|
|
unknown
|
|
>;
|
|
const step1 = (readSetting("step1_config") || {}) as Record<string, unknown>;
|
|
const endpointOverride =
|
|
typeof config.endpoint === "string" && config.endpoint.trim()
|
|
? config.endpoint.trim()
|
|
: null;
|
|
const serverHost =
|
|
typeof step1.serverHost === "string" && step1.serverHost.trim()
|
|
? step1.serverHost.trim()
|
|
: null;
|
|
const endpoint = endpointOverride || buildEndpoint(serverHost);
|
|
const disabled = config.enabled === false;
|
|
const batchSize =
|
|
typeof config.batchSize === "number" && config.batchSize > 0
|
|
? Math.min(Math.floor(config.batchSize), 200)
|
|
: DEFAULT_BATCH_SIZE;
|
|
|
|
return {
|
|
enabled: !disabled,
|
|
endpoint,
|
|
clientKey: readClientKey(serverHost),
|
|
authToken: readAuthToken(serverHost),
|
|
batchSize,
|
|
};
|
|
}
|
|
|
|
function buildEndpoint(serverHost: string | null): string | null {
|
|
if (!serverHost) return null;
|
|
const normalized = serverHost.startsWith("http")
|
|
? serverHost
|
|
: `https://${serverHost}`;
|
|
return `${normalized.replace(/\/+$/, "")}${DEFAULT_SYNC_PATH}`;
|
|
}
|
|
|
|
function missingSyncCredentialLabels(
|
|
config: DigitalEmployeeSyncConfig,
|
|
): string[] {
|
|
return [
|
|
config.clientKey ? "" : "客户端 Key",
|
|
config.authToken ? "" : "登录 token",
|
|
].filter((value): value is string => Boolean(value));
|
|
}
|
|
|
|
function readClientKey(serverHost: string | null): string | null {
|
|
const username = readSetting("auth.username");
|
|
const globalKey = readSetting("auth.saved_key");
|
|
if (serverHost && typeof username === "string" && username.trim()) {
|
|
const domain = serverHost.replace(/^https?:\/\//, "").replace(/\/+$/, "");
|
|
const domainKey = readSetting(`auth.saved_keys.${domain}_${username}`);
|
|
if (typeof domainKey === "string" && domainKey.trim()) return domainKey;
|
|
}
|
|
return typeof globalKey === "string" && globalKey.trim() ? globalKey : null;
|
|
}
|
|
|
|
function readAuthToken(serverHost: string | null): string | null {
|
|
const oneShotToken = readSetting("auth.token");
|
|
if (typeof oneShotToken === "string" && oneShotToken.trim()) {
|
|
return oneShotToken;
|
|
}
|
|
if (serverHost) {
|
|
const domainToken = readSetting(getDomainTokenKey(serverHost));
|
|
if (typeof domainToken === "string" && domainToken.trim()) {
|
|
return domainToken;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function readPendingOutbox(limit: number, force: boolean): OutboxRow[] {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return [];
|
|
const candidateLimit = force
|
|
? limit
|
|
: limit * FAILED_RETRY_CANDIDATE_MULTIPLIER;
|
|
const rows = db
|
|
.prepare(
|
|
`
|
|
SELECT id, entity_type, entity_id, operation, payload, attempts, last_attempt_at
|
|
FROM digital_sync_outbox
|
|
WHERE status IN ('pending', 'failed')
|
|
ORDER BY created_at ASC
|
|
LIMIT ?
|
|
`,
|
|
)
|
|
.all(candidateLimit) as OutboxRow[];
|
|
if (force) return rows.slice(0, limit);
|
|
return rows.filter(isOutboxRowDueForRetry).slice(0, limit);
|
|
}
|
|
|
|
function isOutboxRowDueForRetry(row: OutboxRow): boolean {
|
|
if (!row.last_attempt_at) return true;
|
|
const lastAttemptAt = Date.parse(row.last_attempt_at);
|
|
if (!Number.isFinite(lastAttemptAt)) return true;
|
|
return Date.now() >= lastAttemptAt + retryDelayMs(row.attempts);
|
|
}
|
|
|
|
function retryDelayMs(attempts: number): number {
|
|
const exponent = Math.max(0, Math.min(attempts - 1, 10));
|
|
return Math.min(DEFAULT_INTERVAL_MS * 2 ** exponent, MAX_RETRY_DELAY_MS);
|
|
}
|
|
|
|
async function postOutbox(
|
|
config: DigitalEmployeeSyncConfig,
|
|
rows: OutboxRow[],
|
|
): Promise<SyncResponse> {
|
|
const controller = new AbortController();
|
|
const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
|
|
try {
|
|
const response = await fetch(config.endpoint!, {
|
|
method: "POST",
|
|
headers: {
|
|
"Content-Type": "application/json",
|
|
Authorization: `Bearer ${config.authToken}`,
|
|
"X-Qiming-Client-Key": config.clientKey!,
|
|
},
|
|
body: JSON.stringify({
|
|
client_key: config.clientKey,
|
|
device_id: getDeviceId(),
|
|
contract_version: DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION,
|
|
items: rows.map(buildSyncItem),
|
|
}),
|
|
signal: controller.signal,
|
|
});
|
|
const text = await response.text();
|
|
const body = text ? (JSON.parse(text) as SyncResponse) : {};
|
|
if (!response.ok) {
|
|
throw new Error(body.message || `HTTP ${response.status}`);
|
|
}
|
|
return body;
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
function buildSyncItem(row: OutboxRow): Record<string, unknown> {
|
|
const payload = parsePayload(row.payload);
|
|
const summary = readEntitySummary(row.entity_type, row.entity_id);
|
|
return {
|
|
outbox_id: row.id,
|
|
entity_type: row.entity_type,
|
|
entity_id: row.entity_id,
|
|
operation: row.operation,
|
|
contract_version: DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION,
|
|
entity_summary: summary,
|
|
business_view: buildBusinessView(row, payload, summary),
|
|
payload,
|
|
};
|
|
}
|
|
|
|
function buildBusinessView(
|
|
row: OutboxRow,
|
|
payload: unknown,
|
|
summary: DigitalEmployeeSyncEntitySummary | null,
|
|
): Record<string, unknown> {
|
|
const record = objectRecord(payload) ?? {};
|
|
const base = {
|
|
entity_type: row.entity_type,
|
|
local_id: row.entity_id,
|
|
operation: row.operation,
|
|
title: stringField(record.title) || summary?.title || null,
|
|
status: stringField(record.status) || summary?.status || null,
|
|
summary: stringField(record.summary) || summary?.summary || null,
|
|
};
|
|
|
|
switch (row.entity_type) {
|
|
case "plan":
|
|
return {
|
|
...base,
|
|
objective: stringField(record.objective) || summary?.summary || null,
|
|
source: stringField(record.source) || "qimingclaw",
|
|
};
|
|
case "task":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.planId ?? record.plan_id) || null,
|
|
assigned_skill_id: stringField(record.assignedSkillId ?? record.assigned_skill_id) || null,
|
|
};
|
|
case "plan_step":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.planId ?? record.plan_id) || null,
|
|
step_id: row.entity_id,
|
|
seq: typeof record.seq === "number" ? record.seq : null,
|
|
depends_on: Array.isArray(record.dependsOn ?? record.depends_on)
|
|
? record.dependsOn ?? record.depends_on
|
|
: [],
|
|
preferred_skill_ids: Array.isArray(record.preferredSkillIds ?? record.preferred_skill_ids)
|
|
? record.preferredSkillIds ?? record.preferred_skill_ids
|
|
: [],
|
|
};
|
|
case "run":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.planId ?? record.plan_id) || null,
|
|
task_id: stringField(record.taskId ?? record.task_id) || null,
|
|
started_at: stringField(record.startedAt ?? record.started_at ?? record.now) || null,
|
|
finished_at: stringField(record.finishedAt ?? record.finished_at) || null,
|
|
};
|
|
case "event":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.plan_id ?? record.planId) || null,
|
|
task_id: stringField(record.task_id ?? record.taskId) || null,
|
|
run_id: stringField(record.run_id ?? record.runId) || null,
|
|
kind: stringField(record.kind) || summary?.status || null,
|
|
message: stringField(record.message) || summary?.summary || null,
|
|
occurred_at: stringField(record.occurred_at ?? record.occurredAt) || null,
|
|
};
|
|
case "artifact":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.planId ?? record.plan_id) || null,
|
|
task_id: stringField(record.taskId ?? record.task_id) || null,
|
|
run_id: stringField(record.runId ?? record.run_id) || null,
|
|
kind: stringField(record.kind) || summary?.status || null,
|
|
name: stringField(record.name) || summary?.title || null,
|
|
uri: stringField(record.uri) || summary?.summary || null,
|
|
};
|
|
case "approval":
|
|
return {
|
|
...base,
|
|
plan_id: stringField(record.planId ?? record.plan_id) || null,
|
|
task_id: stringField(record.taskId ?? record.task_id) || null,
|
|
run_id: stringField(record.runId ?? record.run_id) || null,
|
|
decision: stringField(record.decision) || null,
|
|
};
|
|
case "memory":
|
|
return {
|
|
...base,
|
|
key: stringField(record.key) || row.entity_id,
|
|
category: stringField(record.category) || null,
|
|
source: stringField(record.source) || null,
|
|
content_preview: compactPreview(stringField(record.content) || summary?.summary || ""),
|
|
};
|
|
default:
|
|
return base;
|
|
}
|
|
}
|
|
|
|
function compactPreview(value: string, maxLength = 120): string | null {
|
|
const compact = value.replace(/\s+/g, " ").trim();
|
|
if (!compact) return null;
|
|
return compact.length > maxLength ? `${compact.slice(0, maxLength)}...` : compact;
|
|
}
|
|
|
|
function markRowsSyncing(rows: OutboxRow[], now: string): void {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return;
|
|
const tx = db.transaction(() => {
|
|
const stmt = db.prepare(`
|
|
UPDATE digital_sync_outbox
|
|
SET status = 'syncing', attempts = attempts + 1, last_attempt_at = ?, updated_at = ?
|
|
WHERE id = ?
|
|
`);
|
|
for (const row of rows) stmt.run(now, now, row.id);
|
|
});
|
|
tx();
|
|
}
|
|
|
|
function markRowsSynced(
|
|
acks: SyncAck[],
|
|
now: string,
|
|
sourceRows: OutboxRow[] = [],
|
|
): void {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return;
|
|
const tx = db.transaction(() => {
|
|
const outboxStmt = db.prepare(`
|
|
UPDATE digital_sync_outbox
|
|
SET status = 'synced', error = NULL, updated_at = ?
|
|
WHERE id = ?
|
|
`);
|
|
for (const ack of acks) {
|
|
const outboxId =
|
|
ack.outbox_id ||
|
|
ack.id ||
|
|
sourceRows.find((row) => ackMatchesRow(ack, row))?.id;
|
|
if (!outboxId) continue;
|
|
outboxStmt.run(now, outboxId);
|
|
updateEntitySyncStatus(ack, now);
|
|
}
|
|
});
|
|
tx();
|
|
}
|
|
|
|
function ackMatchesRow(ack: SyncAck, row: OutboxRow): boolean {
|
|
const outboxId = ack.outbox_id || ack.id;
|
|
if (outboxId) return outboxId === row.id;
|
|
return ack.entity_type === row.entity_type && ack.entity_id === row.entity_id;
|
|
}
|
|
|
|
function recordSyncAttempts(
|
|
rows: OutboxRow[],
|
|
config: DigitalEmployeeSyncConfig,
|
|
status: "synced" | "failed",
|
|
error: string | null,
|
|
startedAt: string,
|
|
finishedAt: string,
|
|
): void {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return;
|
|
const tx = db.transaction(() => {
|
|
const stmt = db.prepare(`
|
|
INSERT INTO digital_sync_attempts (
|
|
outbox_id, entity_type, entity_id, operation, attempt_no,
|
|
status, error, endpoint, started_at, finished_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`);
|
|
for (const row of rows) {
|
|
stmt.run(
|
|
row.id,
|
|
row.entity_type,
|
|
row.entity_id,
|
|
row.operation,
|
|
row.attempts + 1,
|
|
status,
|
|
error,
|
|
config.endpoint,
|
|
startedAt,
|
|
finishedAt,
|
|
);
|
|
}
|
|
db.prepare("DELETE FROM digital_sync_attempts WHERE finished_at < ?").run(
|
|
new Date(Date.parse(finishedAt) - ATTEMPT_HISTORY_RETENTION_MS).toISOString(),
|
|
);
|
|
});
|
|
tx();
|
|
}
|
|
|
|
function updateEntitySyncStatus(ack: SyncAck, now: string): void {
|
|
if (!ack.entity_type || !ack.entity_id) return;
|
|
const table = entityTable(ack.entity_type);
|
|
if (!table) return;
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return;
|
|
db.prepare(`
|
|
UPDATE ${table}
|
|
SET sync_status = 'synced',
|
|
remote_id = COALESCE(?, remote_id),
|
|
last_synced_at = ?,
|
|
sync_error = NULL
|
|
WHERE id = ?
|
|
`).run(ack.remote_id || null, now, ack.entity_id);
|
|
}
|
|
|
|
function markRowsFailed(rows: OutboxRow[], now: string, error: string): void {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return;
|
|
const tx = db.transaction(() => {
|
|
const stmt = db.prepare(`
|
|
UPDATE digital_sync_outbox
|
|
SET status = 'failed', error = ?, updated_at = ?
|
|
WHERE id = ?
|
|
`);
|
|
for (const row of rows) {
|
|
stmt.run(error, now, row.id);
|
|
markEntityFailed(row.entity_type, row.entity_id, error);
|
|
}
|
|
});
|
|
tx();
|
|
}
|
|
|
|
function markEntityFailed(entityType: string, entityId: string, error: string): void {
|
|
const table = entityTable(entityType);
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db || !table) return;
|
|
db.prepare(`
|
|
UPDATE ${table}
|
|
SET sync_status = 'failed', sync_error = ?
|
|
WHERE id = ?
|
|
`).run(error, entityId);
|
|
}
|
|
|
|
function countOutbox(status: string): number {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return 0;
|
|
const row = db
|
|
.prepare("SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?")
|
|
.get(status) as { count: number } | undefined;
|
|
return row?.count ?? 0;
|
|
}
|
|
|
|
function readFailureSummary(): DigitalEmployeeSyncFailureSummary {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return { total: 0, dueForRetry: 0, backoff: 0, byEntityType: [] };
|
|
const rows = db
|
|
.prepare(
|
|
`
|
|
SELECT entity_type, attempts, last_attempt_at
|
|
FROM digital_sync_outbox
|
|
WHERE status = 'failed'
|
|
`,
|
|
)
|
|
.all() as Array<{
|
|
entity_type: string;
|
|
attempts: number;
|
|
last_attempt_at: string | null;
|
|
}>;
|
|
const groups = new Map<string, number>();
|
|
let dueForRetry = 0;
|
|
|
|
for (const row of rows) {
|
|
groups.set(row.entity_type, (groups.get(row.entity_type) || 0) + 1);
|
|
const nextRetryAt = nextRetryAtForFailure(row.last_attempt_at, row.attempts);
|
|
if (!nextRetryAt || Date.now() >= Date.parse(nextRetryAt)) {
|
|
dueForRetry += 1;
|
|
}
|
|
}
|
|
|
|
return {
|
|
total: rows.length,
|
|
dueForRetry,
|
|
backoff: rows.length - dueForRetry,
|
|
byEntityType: Array.from(groups.entries())
|
|
.map(([entityType, count]) => ({ entityType, count }))
|
|
.sort((left, right) => right.count - left.count),
|
|
};
|
|
}
|
|
|
|
function readRecentFailures(
|
|
config: DigitalEmployeeSyncConfig,
|
|
limit = 3,
|
|
): DigitalEmployeeSyncFailure[] {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return [];
|
|
const rows = db
|
|
.prepare(
|
|
`
|
|
SELECT id, entity_type, entity_id, operation, attempts,
|
|
last_attempt_at, error, created_at, updated_at
|
|
FROM digital_sync_outbox
|
|
WHERE status = 'failed'
|
|
ORDER BY COALESCE(last_attempt_at, updated_at, created_at) DESC
|
|
LIMIT ?
|
|
`,
|
|
)
|
|
.all(limit) as Array<{
|
|
id: string;
|
|
entity_type: string;
|
|
entity_id: string;
|
|
operation: string;
|
|
attempts: number;
|
|
last_attempt_at: string | null;
|
|
error: string | null;
|
|
created_at: string;
|
|
updated_at: string;
|
|
}>;
|
|
|
|
return rows.map((row) => {
|
|
const nextRetryAt = nextRetryAtForFailure(row.last_attempt_at, row.attempts);
|
|
return {
|
|
id: row.id,
|
|
entityType: row.entity_type,
|
|
entityId: row.entity_id,
|
|
entitySummary: readEntitySummary(row.entity_type, row.entity_id),
|
|
operation: row.operation,
|
|
attempts: row.attempts,
|
|
lastAttemptAt: row.last_attempt_at,
|
|
nextRetryAt,
|
|
dueForRetry: !nextRetryAt || Date.now() >= Date.parse(nextRetryAt),
|
|
managementRecordUrl: buildManagementRecordUrl(config, row),
|
|
attemptHistory: readAttemptHistory(row.id),
|
|
error: row.error,
|
|
createdAt: row.created_at,
|
|
updatedAt: row.updated_at,
|
|
};
|
|
});
|
|
}
|
|
|
|
function readEntitySummary(
|
|
entityType: string,
|
|
entityId: string,
|
|
): DigitalEmployeeSyncEntitySummary | null {
|
|
const table = entityTable(entityType);
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db || !table) return null;
|
|
const select = entitySummarySelect(entityType);
|
|
if (!select) return null;
|
|
const row = db.prepare(select).get(entityId) as Record<string, unknown> | undefined;
|
|
if (!row) return null;
|
|
const payload = parsePayload(String(row.payload ?? "{}"));
|
|
const payloadSummary = payload && typeof payload === "object" && !Array.isArray(payload)
|
|
? payload as Record<string, unknown>
|
|
: {};
|
|
return {
|
|
title: stringField(row.title) || stringField(payloadSummary.title) || null,
|
|
status: stringField(row.status) || stringField(payloadSummary.status) || null,
|
|
summary:
|
|
stringField(row.summary) ||
|
|
stringField(row.objective) ||
|
|
stringField(row.message) ||
|
|
stringField(payloadSummary.objective) ||
|
|
stringField(payloadSummary.message) ||
|
|
stringField(payloadSummary.summary) ||
|
|
null,
|
|
};
|
|
}
|
|
|
|
function entitySummarySelect(entityType: string): string | null {
|
|
switch (entityType) {
|
|
case "plan":
|
|
return "SELECT title, status, objective, payload FROM digital_plans WHERE id = ?";
|
|
case "task":
|
|
return "SELECT title, status, NULL AS objective, payload FROM digital_tasks WHERE id = ?";
|
|
case "plan_step":
|
|
return "SELECT title, status, NULL AS objective, payload FROM digital_plan_steps WHERE id = ?";
|
|
case "schedule":
|
|
return "SELECT name AS title, status, description AS objective, payload FROM digital_schedules WHERE id = ?";
|
|
case "schedule_run":
|
|
return "SELECT id AS title, status, error AS objective, payload FROM digital_schedule_runs WHERE id = ?";
|
|
case "run":
|
|
return "SELECT id AS title, status, NULL AS objective, payload FROM digital_runs WHERE id = ?";
|
|
case "event":
|
|
return "SELECT kind AS title, kind AS status, message, payload FROM digital_events WHERE id = ?";
|
|
case "artifact":
|
|
return "SELECT name AS title, kind AS status, uri AS summary, payload FROM digital_artifacts WHERE id = ?";
|
|
case "approval":
|
|
return "SELECT title, status, NULL AS objective, payload FROM digital_approvals WHERE id = ?";
|
|
case "memory":
|
|
return "SELECT key AS title, status, content AS summary, payload FROM digital_memories WHERE id = ?";
|
|
default:
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function stringField(value: unknown): string {
|
|
return typeof value === "string" && value.trim() ? value.trim() : "";
|
|
}
|
|
|
|
function objectRecord(value: unknown): Record<string, unknown> | null {
|
|
return value && typeof value === "object" && !Array.isArray(value)
|
|
? value as Record<string, unknown>
|
|
: null;
|
|
}
|
|
|
|
function readAttemptHistory(
|
|
outboxId: string,
|
|
limit = 5,
|
|
): DigitalEmployeeSyncAttempt[] {
|
|
const db = getDigitalEmployeeDb();
|
|
if (!db) return [];
|
|
const rows = db
|
|
.prepare(
|
|
`
|
|
SELECT attempt_no, status, error, endpoint, started_at, finished_at
|
|
FROM digital_sync_attempts
|
|
WHERE outbox_id = ?
|
|
ORDER BY finished_at DESC, id DESC
|
|
LIMIT ?
|
|
`,
|
|
)
|
|
.all(outboxId, limit) as Array<{
|
|
attempt_no: number;
|
|
status: string;
|
|
error: string | null;
|
|
endpoint: string | null;
|
|
started_at: string;
|
|
finished_at: string;
|
|
}>;
|
|
|
|
return rows.map((row) => ({
|
|
attemptNo: row.attempt_no,
|
|
status: row.status,
|
|
error: row.error,
|
|
endpoint: row.endpoint,
|
|
startedAt: row.started_at,
|
|
finishedAt: row.finished_at,
|
|
}));
|
|
}
|
|
|
|
function buildManagementRecordUrl(
|
|
config: DigitalEmployeeSyncConfig,
|
|
row: {
|
|
id: string;
|
|
entity_type: string;
|
|
entity_id: string;
|
|
},
|
|
): string | null {
|
|
if (!config.endpoint) return null;
|
|
try {
|
|
const endpoint = new URL(config.endpoint);
|
|
const url = new URL(MANAGEMENT_SYNC_RECORD_PATH, endpoint.origin);
|
|
url.searchParams.set("device_id", getDeviceId());
|
|
url.searchParams.set("entity_type", row.entity_type);
|
|
url.searchParams.set("entity_id", row.entity_id);
|
|
url.searchParams.set("outbox_id", row.id);
|
|
return url.toString();
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function nextRetryAtForFailure(
|
|
lastAttemptAt: string | null,
|
|
attempts: number,
|
|
): string | null {
|
|
if (!lastAttemptAt) return null;
|
|
const lastAttemptTime = Date.parse(lastAttemptAt);
|
|
if (!Number.isFinite(lastAttemptTime)) return null;
|
|
return new Date(lastAttemptTime + retryDelayMs(attempts)).toISOString();
|
|
}
|
|
|
|
function entityTable(entityType: string): string | null {
|
|
switch (entityType) {
|
|
case "plan":
|
|
return "digital_plans";
|
|
case "task":
|
|
return "digital_tasks";
|
|
case "plan_step":
|
|
return "digital_plan_steps";
|
|
case "schedule":
|
|
return "digital_schedules";
|
|
case "schedule_run":
|
|
return "digital_schedule_runs";
|
|
case "run":
|
|
return "digital_runs";
|
|
case "event":
|
|
return "digital_events";
|
|
case "artifact":
|
|
return "digital_artifacts";
|
|
case "approval":
|
|
return "digital_approvals";
|
|
case "memory":
|
|
return "digital_memories";
|
|
default:
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function parsePayload(payload: string): unknown {
|
|
try {
|
|
return JSON.parse(payload);
|
|
} catch {
|
|
return payload;
|
|
}
|
|
}
|
|
|
|
function normalizeError(error: unknown): string {
|
|
if (error instanceof Error) return error.message;
|
|
return String(error);
|
|
}
|