From f8564b631db2d496622a213a81fe0b795b828698 Mon Sep 17 00:00:00 2001 From: baiyanyun Date: Sun, 7 Jun 2026 02:16:04 +0800 Subject: [PATCH] test(client): cover digital employee sync retries --- .../digitalEmployee/syncService.test.ts | 462 ++++++++++++++++++ .../services/digitalEmployee/syncService.ts | 13 +- 2 files changed, 472 insertions(+), 3 deletions(-) create mode 100644 qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts diff --git a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts new file mode 100644 index 00000000..7ee7db83 --- /dev/null +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts @@ -0,0 +1,462 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const mockState = vi.hoisted(() => ({ + db: null as TestDb | null, + settings: new Map(), + fetch: vi.fn(), +})); + +vi.mock("electron-log", () => ({ + default: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }, +})); + +vi.mock("../system/deviceId", () => ({ + getDeviceId: vi.fn(() => "device-001"), +})); + +vi.mock("../../db", () => ({ + getDb: () => mockState.db, + readSetting: (key: string) => mockState.settings.get(key) ?? null, +})); + +describe("digital employee sync service", () => { + beforeEach(() => { + vi.resetModules(); + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-06-07T08:00:00.000Z")); + mockState.fetch = vi.fn(); + vi.stubGlobal("fetch", mockState.fetch); + mockState.settings = new Map([ + ["step1_config", { serverHost: "https://manage.example.com" }], + ["auth.username", "operator"], + ["auth.saved_key", "client-key-001"], + ["auth.tokens.manage.example.com", "login-token-001"], + ]); + mockState.db = new TestDb(); + }); + + afterEach(() => { + mockState.db = null; + vi.unstubAllGlobals(); + vi.useRealTimers(); + }); + + it("skips flushing when management credentials are incomplete", async () => { + mockState.settings.delete("auth.saved_key"); + insertOutbox("plan:upsert:plan-1", "plan", "plan-1"); + + const { flushDigitalEmployeeSyncOutbox } = await import("./syncService"); + const status = await flushDigitalEmployeeSyncOutbox({ force: true }); + + expect(mockState.fetch).not.toHaveBeenCalled(); + expect(status.missingCredentials).toEqual(["客户端 Key"]); + expect(status.pending).toBe(1); + expect(status.failed).toBe(0); + expect(readOutbox("plan:upsert:plan-1")).toMatchObject({ + status: "pending", + attempts: 0, + }); + }); + + it("marks entity-only acknowledgements as synced", async () => { + insertPlan("plan-1"); + insertOutbox("plan:upsert:plan-1", "plan", "plan-1"); + mockState.fetch.mockResolvedValue( + jsonResponse({ + success: true, + acked: [ + { + entity_type: "plan", + entity_id: "plan-1", + remote_id: "remote-plan-1", + }, + ], + }), + ); + + const { flushDigitalEmployeeSyncOutbox } = await import("./syncService"); + const status = await flushDigitalEmployeeSyncOutbox({ force: true }); + + expect(status.pending).toBe(0); + expect(status.failed).toBe(0); + expect(readOutbox("plan:upsert:plan-1")).toMatchObject({ + status: "synced", + error: null, + attempts: 1, + }); + expect(readPlan("plan-1")).toMatchObject({ + sync_status: "synced", + remote_id: "remote-plan-1", + sync_error: null, + }); + expect(readAttempts()).toEqual([ + expect.objectContaining({ + outbox_id: "plan:upsert:plan-1", + attempt_no: 1, + status: "synced", + error: null, + }), + ]); + }); + + it("keeps unacknowledged rows retryable after a partial backend ack", async () => { + insertPlan("plan-1"); + insertPlan("plan-2"); + insertOutbox("plan:upsert:plan-1", "plan", "plan-1"); + insertOutbox("plan:upsert:plan-2", "plan", "plan-2"); + mockState.fetch.mockResolvedValue( + jsonResponse({ + success: true, + acked: [ + { + outbox_id: "plan:upsert:plan-1", + entity_type: "plan", + entity_id: "plan-1", + }, + ], + }), + ); + + const { flushDigitalEmployeeSyncOutbox } = await import("./syncService"); + const status = await flushDigitalEmployeeSyncOutbox({ force: true }); + + expect(status.lastSyncError).toBe( + "management sync response did not acknowledge item", + ); + expect(status.failed).toBe(1); + expect(status.failureSummary).toMatchObject({ + total: 1, + dueForRetry: 0, + backoff: 1, + byEntityType: [{ entityType: "plan", count: 1 }], + }); + expect(readOutbox("plan:upsert:plan-1")).toMatchObject({ + status: "synced", + }); + expect(readOutbox("plan:upsert:plan-2")).toMatchObject({ + status: "failed", + attempts: 1, + error: "management sync response did not acknowledge item", + }); + expect(status.recentFailures[0]).toMatchObject({ + id: "plan:upsert:plan-2", + dueForRetry: false, + managementRecordUrl: + "https://manage.example.com/system/content/content-digital-employee-sync?client_key=client-key-001&entity_type=plan&entity_id=plan-2&outbox_id=plan%3Aupsert%3Aplan-2", + attemptHistory: [ + expect.objectContaining({ + attemptNo: 1, + status: "failed", + error: "management sync response did not acknowledge item", + endpoint: "https://manage.example.com/api/digital-employee/sync/outbox", + }), + ], + }); + }); +}); + +function insertPlan(id: string): void { + mockState.db?.plans.set(id, { + id, + remote_id: null, + sync_status: "pending", + sync_error: null, + last_synced_at: null, + }); +} + +function insertOutbox(id: string, entityType: string, entityId: string): void { + mockState.db?.outbox.set(id, { + id, + entity_type: entityType, + entity_id: entityId, + operation: "upsert", + payload: '{"ok":true}', + status: "pending", + attempts: 0, + last_attempt_at: null, + error: null, + created_at: "2026-06-07T07:55:00.000Z", + updated_at: "2026-06-07T07:55:00.000Z", + }); +} + +function readOutbox(id: string): Record | undefined { + return mockState.db?.outbox.get(id); +} + +function readPlan(id: string): Record | undefined { + return mockState.db?.plans.get(id); +} + +function readAttempts(): Array> { + return mockState.db?.attempts.map((attempt) => ({ ...attempt })) ?? []; +} + +function jsonResponse(body: unknown): Response { + return { + ok: true, + status: 200, + text: vi.fn().mockResolvedValue(JSON.stringify(body)), + } as unknown as Response; +} + +interface TestOutboxRow { + id: string; + entity_type: string; + entity_id: string; + operation: string; + payload: string; + status: string; + attempts: number; + last_attempt_at: string | null; + error: string | null; + created_at: string; + updated_at: string; +} + +interface TestPlanRow { + id: string; + remote_id: string | null; + sync_status: string; + sync_error: string | null; + last_synced_at: string | null; +} + +interface TestAttemptRow { + id: number; + outbox_id: string; + entity_type: string; + entity_id: string; + operation: string; + attempt_no: number; + status: string; + error: string | null; + endpoint: string | null; + started_at: string; + finished_at: string; +} + +class TestDb { + outbox = new Map(); + plans = new Map(); + attempts: TestAttemptRow[] = []; + private nextAttemptId = 1; + + transaction(callback: () => T): () => T { + return callback; + } + + prepare(sql: string): { + all: (...args: unknown[]) => unknown[]; + get: (...args: unknown[]) => unknown; + run: (...args: unknown[]) => unknown; + } { + const normalized = sql.replace(/\s+/g, " ").trim(); + return { + all: (...args: unknown[]) => this.all(normalized, args), + get: (...args: unknown[]) => this.get(normalized, args), + run: (...args: unknown[]) => this.run(normalized, args), + }; + } + + private all(sql: string, args: unknown[]): unknown[] { + if ( + sql.includes("FROM digital_sync_outbox") && + sql.includes("status IN ('pending', 'failed')") + ) { + const limit = Number(args[0]); + return Array.from(this.outbox.values()) + .filter((row) => row.status === "pending" || row.status === "failed") + .sort((left, right) => left.created_at.localeCompare(right.created_at)) + .slice(0, limit) + .map(copyRow); + } + + if ( + sql.includes("FROM digital_sync_outbox") && + sql.includes("WHERE status = 'failed'") && + sql.includes("SELECT entity_type") + ) { + return Array.from(this.outbox.values()) + .filter((row) => row.status === "failed") + .map((row) => ({ + entity_type: row.entity_type, + attempts: row.attempts, + last_attempt_at: row.last_attempt_at, + })); + } + + if ( + sql.includes("FROM digital_sync_outbox") && + sql.includes("WHERE status = 'failed'") && + sql.includes("ORDER BY COALESCE") + ) { + const limit = Number(args[0]); + return Array.from(this.outbox.values()) + .filter((row) => row.status === "failed") + .sort((left, right) => + recentFailureSortKey(right).localeCompare(recentFailureSortKey(left)), + ) + .slice(0, limit) + .map(copyRow); + } + + if (sql.includes("FROM digital_sync_attempts")) { + const [outboxId, limit] = args as [string, number]; + return this.attempts + .filter((row) => row.outbox_id === outboxId) + .sort((left, right) => { + const byTime = right.finished_at.localeCompare(left.finished_at); + return byTime || right.id - left.id; + }) + .slice(0, limit) + .map((row) => ({ + attempt_no: row.attempt_no, + status: row.status, + error: row.error, + endpoint: row.endpoint, + started_at: row.started_at, + finished_at: row.finished_at, + })); + } + + throw new Error(`Unhandled all query: ${sql}`); + } + + private get(sql: string, args: unknown[]): unknown { + if ( + sql.includes("COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?") + ) { + const [status] = args; + return { + count: Array.from(this.outbox.values()).filter( + (row) => row.status === status, + ).length, + }; + } + + throw new Error(`Unhandled get query: ${sql}`); + } + + private run(sql: string, args: unknown[]): unknown { + if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'syncing'")) { + const [lastAttemptAt, updatedAt, id] = args as [string, string, string]; + const row = this.outbox.get(id); + if (row) { + row.status = "syncing"; + row.attempts += 1; + row.last_attempt_at = lastAttemptAt; + row.updated_at = updatedAt; + } + return { changes: row ? 1 : 0 }; + } + + if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'synced'")) { + const [updatedAt, id] = args as [string, string]; + const row = this.outbox.get(id); + if (row) { + row.status = "synced"; + row.error = null; + row.updated_at = updatedAt; + } + return { changes: row ? 1 : 0 }; + } + + if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'failed'")) { + const [error, updatedAt, id] = args as [string, string, string]; + const row = this.outbox.get(id); + if (row) { + row.status = "failed"; + row.error = error; + row.updated_at = updatedAt; + } + return { changes: row ? 1 : 0 }; + } + + if (sql.startsWith("INSERT INTO digital_sync_attempts")) { + const [ + outboxId, + entityType, + entityId, + operation, + attemptNo, + status, + error, + endpoint, + startedAt, + finishedAt, + ] = args as [ + string, + string, + string, + string, + number, + string, + string | null, + string | null, + string, + string, + ]; + this.attempts.push({ + id: this.nextAttemptId, + outbox_id: outboxId, + entity_type: entityType, + entity_id: entityId, + operation, + attempt_no: attemptNo, + status, + error, + endpoint, + started_at: startedAt, + finished_at: finishedAt, + }); + this.nextAttemptId += 1; + return { changes: 1 }; + } + + if (sql.startsWith("DELETE FROM digital_sync_attempts WHERE finished_at < ?")) { + const [cutoff] = args as [string]; + const before = this.attempts.length; + this.attempts = this.attempts.filter((row) => row.finished_at >= cutoff); + return { changes: before - this.attempts.length }; + } + + if (sql.startsWith("UPDATE digital_plans SET sync_status = 'synced'")) { + const [remoteId, lastSyncedAt, id] = args as [string | null, string, string]; + const row = this.plans.get(id); + if (row) { + row.sync_status = "synced"; + row.remote_id = remoteId ?? row.remote_id; + row.last_synced_at = lastSyncedAt; + row.sync_error = null; + } + return { changes: row ? 1 : 0 }; + } + + if (sql.startsWith("UPDATE digital_plans SET sync_status = 'failed'")) { + const [syncError, id] = args as [string, string]; + const row = this.plans.get(id); + if (row) { + row.sync_status = "failed"; + row.sync_error = syncError; + } + return { changes: row ? 1 : 0 }; + } + + throw new Error(`Unhandled run query: ${sql}`); + } +} + +function copyRow(row: T): T { + return { ...row }; +} + +function recentFailureSortKey(row: TestOutboxRow): string { + return row.last_attempt_at || row.updated_at || row.created_at; +} diff --git a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts index b4bf307f..5ec461a6 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts @@ -183,7 +183,7 @@ export async function flushDigitalEmployeeSyncOutbox( : []; recordSyncAttempts(syncedRows, config, "synced", null, startedAt, finishedAt); if (acked.length > 0) { - markRowsSynced(acked, finishedAt); + markRowsSynced(acked, finishedAt, rows); } else { markRowsSynced( rows.map((row) => ({ @@ -380,7 +380,11 @@ function markRowsSyncing(rows: OutboxRow[], now: string): void { tx(); } -function markRowsSynced(acks: SyncAck[], now: string): void { +function markRowsSynced( + acks: SyncAck[], + now: string, + sourceRows: OutboxRow[] = [], +): void { const db = getDb(); if (!db) return; const tx = db.transaction(() => { @@ -390,7 +394,10 @@ function markRowsSynced(acks: SyncAck[], now: string): void { WHERE id = ? `); for (const ack of acks) { - const outboxId = ack.outbox_id || ack.id; + const outboxId = + ack.outbox_id || + ack.id || + sourceRows.find((row) => ackMatchesRow(ack, row))?.id; if (!outboxId) continue; outboxStmt.run(now, outboxId); updateEntitySyncStatus(ack, now);