feat(client): sync digital employee outbox
This commit is contained in:
baiyanyun
@@ -0,0 +1,363 @@
|
||||
import log from "electron-log";
|
||||
import { getDeviceId } from "../system/deviceId";
|
||||
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;
|
||||
|
||||
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;
|
||||
batchSize: number;
|
||||
}
|
||||
|
||||
interface OutboxRow {
|
||||
id: string;
|
||||
entity_type: string;
|
||||
entity_id: string;
|
||||
operation: string;
|
||||
payload: string;
|
||||
attempts: number;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
export async function flushDigitalEmployeeSyncOutbox(): Promise<DigitalEmployeeSyncStatus> {
|
||||
if (syncing) return getDigitalEmployeeSyncStatus();
|
||||
|
||||
const config = resolveSyncConfig();
|
||||
if (!config.enabled || !config.endpoint || !config.clientKey) {
|
||||
lastSyncError = null;
|
||||
return getDigitalEmployeeSyncStatus();
|
||||
}
|
||||
|
||||
const rows = readPendingOutbox(config.batchSize);
|
||||
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),
|
||||
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 readPendingOutbox(limit: number): OutboxRow[] {
|
||||
const db = getDb();
|
||||
if (!db) return [];
|
||||
return db
|
||||
.prepare(
|
||||
`
|
||||
SELECT id, entity_type, entity_id, operation, payload, attempts
|
||||
FROM digital_sync_outbox
|
||||
WHERE status IN ('pending', 'failed')
|
||||
ORDER BY created_at ASC
|
||||
LIMIT ?
|
||||
`,
|
||||
)
|
||||
.all(limit) as OutboxRow[];
|
||||
}
|
||||
|
||||
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.clientKey}`,
|
||||
"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 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);
|
||||
}
|
||||
Reference in New Issue
Block a user