456 lines
13 KiB
TypeScript
456 lines
13 KiB
TypeScript
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 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;
|
|
|
|
let syncTimer: ReturnType<typeof setInterval> | 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 DigitalEmployeeSyncFailure {
|
|
id: string;
|
|
entityType: string;
|
|
entityId: string;
|
|
operation: string;
|
|
attempts: number;
|
|
lastAttemptAt: string | null;
|
|
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;
|
|
pending: number;
|
|
failed: number;
|
|
lastSyncAt: string | null;
|
|
lastSyncError: string | null;
|
|
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,
|
|
pending: countOutbox("pending"),
|
|
failed: countOutbox("failed"),
|
|
lastSyncAt,
|
|
lastSyncError,
|
|
recentFailures: readRecentFailures(),
|
|
};
|
|
}
|
|
|
|
export async function flushDigitalEmployeeSyncOutbox(
|
|
options: { force?: boolean } = {},
|
|
): Promise<DigitalEmployeeSyncStatus> {
|
|
if (syncing) return getDigitalEmployeeSyncStatus();
|
|
|
|
const config = resolveSyncConfig();
|
|
if (!config.enabled || !config.endpoint || !config.clientKey || !config.authToken) {
|
|
lastSyncError = null;
|
|
return getDigitalEmployeeSyncStatus();
|
|
}
|
|
|
|
const rows = readPendingOutbox(config.batchSize, options.force === true);
|
|
if (rows.length === 0) {
|
|
lastSyncError = null;
|
|
return getDigitalEmployeeSyncStatus();
|
|
}
|
|
|
|
syncing = true;
|
|
const now = new Date().toISOString();
|
|
markRowsSyncing(rows, now);
|
|
|
|
try {
|
|
const response = await postOutbox(config, rows);
|
|
const acked = response.data?.acked || response.acked || [];
|
|
if (response.success === false) {
|
|
throw new Error(response.message || "digital employee sync rejected");
|
|
}
|
|
if (acked.length > 0) {
|
|
markRowsSynced(acked, now);
|
|
} else {
|
|
markRowsSynced(
|
|
rows.map((row) => ({
|
|
outbox_id: row.id,
|
|
entity_type: row.entity_type,
|
|
entity_id: row.entity_id,
|
|
})),
|
|
now,
|
|
);
|
|
}
|
|
lastSyncAt = now;
|
|
lastSyncError = null;
|
|
} catch (error) {
|
|
lastSyncError = normalizeError(error);
|
|
markRowsFailed(rows, now, 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<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 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<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(),
|
|
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): 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;
|
|
if (!outboxId) continue;
|
|
outboxStmt.run(now, outboxId);
|
|
updateEntitySyncStatus(ack, now);
|
|
}
|
|
});
|
|
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 readRecentFailures(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) => ({
|
|
id: row.id,
|
|
entityType: row.entity_type,
|
|
entityId: row.entity_id,
|
|
operation: row.operation,
|
|
attempts: row.attempts,
|
|
lastAttemptAt: row.last_attempt_at,
|
|
error: row.error,
|
|
createdAt: row.created_at,
|
|
updatedAt: row.updated_at,
|
|
}));
|
|
}
|
|
|
|
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);
|
|
}
|