import log from "electron-log"; import { getDeviceId } from "../system/deviceId"; import { getDomainTokenKey } from "@shared/utils/domain"; import { 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 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 | null = null; let syncing = false; let lastSyncAt: string | null = null; let lastSyncError: string | null = null; 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; operation: string; attempts: number; lastAttemptAt: string | null; nextRetryAt: string | null; dueForRetry: boolean; managementRecordUrl: string | null; attemptHistory: DigitalEmployeeSyncAttempt[]; error: string | null; createdAt: string; updatedAt: string; } interface SyncAck { outbox_id?: string; id?: string; entity_type?: string; entity_id?: string; remote_id?: string; } interface SyncResponse { success?: boolean; 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 { 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 || []; if (response.success === false) { throw new Error(response.message || "digital employee sync rejected"); } 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 resolveSyncConfig(): DigitalEmployeeSyncConfig { const config = (readSetting("digital_employee_sync_config") || {}) as Record< string, unknown >; const step1 = (readSetting("step1_config") || {}) as Record; 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 = getDb(); 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 { 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(), items: rows.map((row) => ({ outbox_id: row.id, entity_type: row.entity_type, entity_id: row.entity_id, operation: row.operation, payload: parsePayload(row.payload), })), }), 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 markRowsSyncing(rows: OutboxRow[], now: string): void { const db = getDb(); 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 = getDb(); 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 = getDb(); 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 = getDb(); 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 = getDb(); 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 = getDb(); 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 = getDb(); 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 = getDb(); 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(); 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 = getDb(); 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, 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 readAttemptHistory( outboxId: string, limit = 5, ): DigitalEmployeeSyncAttempt[] { const db = getDb(); 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 "run": return "digital_runs"; case "event": return "digital_events"; case "artifact": return "digital_artifacts"; case "approval": return "digital_approvals"; 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); }