byy/qiming
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(client): ensure digital employee schema

Browse Source
This commit is contained in:
baiyanyun
2026-06-07 18:39:19 +08:00
parent 6deea42de1
commit 5e76828afe
6 changed files with 196 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
qimingclaw/crates/agent-electron-client/src/main/db.ts
View File

@@ -10,9 +10,149 @@ const dbPath = path.join(qimingHome, `${APP_NAME_IDENTIFIER}.db`);
let db: Database.Database | null = null;
const DIGITAL_EMPLOYEE_SCHEMA_SQL = `
CREATE TABLE IF NOT EXISTS digital_plans (
id TEXT PRIMARY KEY,
remote_id TEXT,
title TEXT NOT NULL,
objective TEXT,
status TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'local',
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_tasks (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
title TEXT NOT NULL,
status TEXT NOT NULL,
assigned_skill_id TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id)
);
CREATE TABLE IF NOT EXISTS digital_runs (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
status TEXT NOT NULL,
started_at TEXT,
finished_at TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id),
FOREIGN KEY(task_id) REFERENCES digital_tasks(id)
);
CREATE TABLE IF NOT EXISTS digital_events (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
kind TEXT NOT NULL,
message TEXT NOT NULL,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
occurred_at TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id),
FOREIGN KEY(task_id) REFERENCES digital_tasks(id),
FOREIGN KEY(run_id) REFERENCES digital_runs(id)
);
CREATE TABLE IF NOT EXISTS digital_artifacts (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
kind TEXT NOT NULL,
name TEXT,
uri TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_approvals (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
title TEXT NOT NULL,
status TEXT NOT NULL,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_sync_outbox (
id TEXT PRIMARY KEY,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
operation TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
last_attempt_at TEXT,
error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_sync_attempts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
outbox_id TEXT NOT NULL,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
operation TEXT NOT NULL,
attempt_no INTEGER NOT NULL,
status TEXT NOT NULL,
error TEXT,
endpoint TEXT,
started_at TEXT NOT NULL,
finished_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_digital_plans_sync ON digital_plans(sync_status);
CREATE INDEX IF NOT EXISTS idx_digital_tasks_plan ON digital_tasks(plan_id);
CREATE INDEX IF NOT EXISTS idx_digital_runs_task ON digital_runs(task_id);
CREATE INDEX IF NOT EXISTS idx_digital_events_plan ON digital_events(plan_id, occurred_at);
CREATE INDEX IF NOT EXISTS idx_digital_sync_outbox_status ON digital_sync_outbox(status, created_at);
CREATE INDEX IF NOT EXISTS idx_digital_sync_attempts_outbox ON digital_sync_attempts(outbox_id, finished_at);
`;
let digitalEmployeeSchemaEnsured = false;
export function initDatabase(): void {
try {
db = new Database(dbPath);
digitalEmployeeSchemaEnsured = false;
log.info('Database initialized at:', dbPath);
// Create tables
@@ -21,148 +161,27 @@ export function initDatabase(): void {
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_plans (
id TEXT PRIMARY KEY,
remote_id TEXT,
title TEXT NOT NULL,
objective TEXT,
status TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'local',
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_tasks (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
title TEXT NOT NULL,
status TEXT NOT NULL,
assigned_skill_id TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id)
);
CREATE TABLE IF NOT EXISTS digital_runs (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
status TEXT NOT NULL,
started_at TEXT,
finished_at TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id),
FOREIGN KEY(task_id) REFERENCES digital_tasks(id)
);
CREATE TABLE IF NOT EXISTS digital_events (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
kind TEXT NOT NULL,
message TEXT NOT NULL,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
occurred_at TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(plan_id) REFERENCES digital_plans(id),
FOREIGN KEY(task_id) REFERENCES digital_tasks(id),
FOREIGN KEY(run_id) REFERENCES digital_runs(id)
);
CREATE TABLE IF NOT EXISTS digital_artifacts (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
kind TEXT NOT NULL,
name TEXT,
uri TEXT,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_approvals (
id TEXT PRIMARY KEY,
remote_id TEXT,
plan_id TEXT,
task_id TEXT,
run_id TEXT,
title TEXT NOT NULL,
status TEXT NOT NULL,
payload TEXT NOT NULL DEFAULT '{}',
sync_status TEXT NOT NULL DEFAULT 'pending',
last_synced_at TEXT,
sync_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_sync_outbox (
id TEXT PRIMARY KEY,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
operation TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
last_attempt_at TEXT,
error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS digital_sync_attempts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
outbox_id TEXT NOT NULL,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
operation TEXT NOT NULL,
attempt_no INTEGER NOT NULL,
status TEXT NOT NULL,
error TEXT,
endpoint TEXT,
started_at TEXT NOT NULL,
finished_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_digital_plans_sync ON digital_plans(sync_status);
CREATE INDEX IF NOT EXISTS idx_digital_tasks_plan ON digital_tasks(plan_id);
CREATE INDEX IF NOT EXISTS idx_digital_runs_task ON digital_runs(task_id);
CREATE INDEX IF NOT EXISTS idx_digital_events_plan ON digital_events(plan_id, occurred_at);
CREATE INDEX IF NOT EXISTS idx_digital_sync_outbox_status ON digital_sync_outbox(status, created_at);
CREATE INDEX IF NOT EXISTS idx_digital_sync_attempts_outbox ON digital_sync_attempts(outbox_id, finished_at);
`);
ensureDigitalEmployeeSchema();
log.info('Database tables created');
} catch (error) {
log.error('Database initialization failed:', error);
}
}
export function ensureDigitalEmployeeSchema(): boolean {
if (!db) return false;
if (digitalEmployeeSchemaEnsured) return true;
try {
db.exec(DIGITAL_EMPLOYEE_SCHEMA_SQL);
digitalEmployeeSchemaEnsured = true;
return true;
} catch (error) {
log.error('Digital employee schema initialization failed:', error);
return false;
}
}
export function getDb(): Database.Database | null {
return db;
}
@@ -171,6 +190,7 @@ export function closeDb(): void {
if (db) {
db.close();
db = null;
digitalEmployeeSchemaEnsured = false;
log.info('[App] Database closed');
}
}

1
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.test.ts
View File

@@ -6,6 +6,7 @@ const mockState = vi.hoisted(() => ({
}));
vi.mock("../../db", () => ({
ensureDigitalEmployeeSchema: () => true,
getDb: () => mockState.db,
readSetting: (key: string) => mockState.settings.get(key) ?? null,
writeSetting: (key: string, value: unknown) => mockState.settings.set(key, value),

37
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts
View File

@@ -1,4 +1,4 @@
import { getDb, readSetting, writeSetting } from "../../db";
import { ensureDigitalEmployeeSchema, getDb, readSetting, writeSetting } from "../../db";
const STATE_KEY = "digital_employee_state_v1";
const UI_STATE_KEY = "digital_employee_ui_state_v1";
@@ -228,6 +228,11 @@ function defaultUiState(): DigitalEmployeeUiState {
};
}
function getDigitalEmployeeDb(): ReturnType<typeof getDb> {
if (!ensureDigitalEmployeeSchema()) return null;
return getDb();
}
function normalizeStringMap(value: unknown): Record<string, string> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
return Object.fromEntries(
@@ -361,7 +366,7 @@ function recordUiApprovalDecision(update: DigitalEmployeeUiStateUpdate): void {
const decision = stringValue(metadata?.decision);
if (!decisionId || !decision) return;
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const existing = db.prepare(`
SELECT id, plan_id, task_id, run_id, title, status, payload, created_at
@@ -410,7 +415,7 @@ function approvalDecisionStatus(decision: string): string {
export function readDigitalEmployeeRuntimeRecords(
limit = 120,
): DigitalEmployeeRuntimeRecords {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) {
return {
generatedAt: new Date().toISOString(),
@@ -505,7 +510,7 @@ export function recordDigitalEmployeeChatReceived(
const ids = chatIds(request);
const title = chatTitle(request);
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return ids;
const tx = db.transaction(() => {
@@ -566,7 +571,7 @@ export function recordDigitalEmployeeChatResult(
});
const status = result.success ? "completed" : "failed";
const title = chatTitle(request);
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
@@ -627,7 +632,7 @@ export function recordDigitalEmployeeRuntimeEvent(
});
const status = event.status ?? "running";
const title = event.message || event.kind;
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
@@ -680,7 +685,7 @@ function persistSnapshotTables(
snapshot: DigitalEmployeeSnapshot,
event: DigitalEmployeeStateEvent | null,
): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const now = snapshot.generatedAt;
@@ -874,7 +879,7 @@ function toApprovalRecord(
}
function countPendingSyncItems(): number {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return 0;
const row = db.prepare(`
SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = 'pending'
@@ -893,7 +898,7 @@ function markOutbox(
payload: unknown,
now: string,
): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_sync_outbox (
@@ -923,7 +928,7 @@ function upsertPlan(input: {
payload: unknown;
now: string;
}): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_plans (
@@ -960,7 +965,7 @@ function upsertTask(input: {
payload: unknown;
now: string;
}): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_tasks (
@@ -998,7 +1003,7 @@ function upsertRun(input: {
payload: unknown;
now: string;
}): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_runs (
@@ -1028,7 +1033,7 @@ function upsertRun(input: {
}
function finishRun(runId: string, status: string, now: string): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
UPDATE digital_runs
@@ -1048,7 +1053,7 @@ function upsertArtifact(input: {
payload: unknown;
now: string;
}): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_artifacts (
@@ -1090,7 +1095,7 @@ function upsertApproval(input: {
payload: unknown;
now: string;
}): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_approvals (
@@ -1127,7 +1132,7 @@ function insertEvent(
runId: string,
payload: unknown,
): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT OR IGNORE INTO digital_events (

1
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts
View File

@@ -20,6 +20,7 @@ vi.mock("../system/deviceId", () => ({
}));
vi.mock("../../db", () => ({
ensureDigitalEmployeeSchema: () => true,
getDb: () => mockState.db,
readSetting: (key: string) => mockState.settings.get(key) ?? null,
}));

29
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts
View File

@@ -1,7 +1,7 @@
import log from "electron-log";
import { getDeviceId } from "../system/deviceId";
import { getDomainTokenKey } from "@shared/utils/domain";
import { getDb, readSetting } from "../../db";
import { ensureDigitalEmployeeSchema, getDb, readSetting } from "../../db";
const DEFAULT_SYNC_PATH = "/api/digital-employee/sync/outbox";
const MANAGEMENT_SYNC_RECORD_PATH =
@@ -18,6 +18,11 @@ 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;
@@ -296,7 +301,7 @@ function readAuthToken(serverHost: string | null): string | null {
}
function readPendingOutbox(limit: number, force: boolean): OutboxRow[] {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return [];
const candidateLimit = force
? limit
@@ -367,7 +372,7 @@ async function postOutbox(
}
function markRowsSyncing(rows: OutboxRow[], now: string): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
const stmt = db.prepare(`
@@ -385,7 +390,7 @@ function markRowsSynced(
now: string,
sourceRows: OutboxRow[] = [],
): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
const outboxStmt = db.prepare(`
@@ -420,7 +425,7 @@ function recordSyncAttempts(
startedAt: string,
finishedAt: string,
): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
const stmt = db.prepare(`
@@ -454,7 +459,7 @@ 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();
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
UPDATE ${table}
@@ -467,7 +472,7 @@ function updateEntitySyncStatus(ack: SyncAck, now: string): void {
}
function markRowsFailed(rows: OutboxRow[], now: string, error: string): void {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
const stmt = db.prepare(`
@@ -485,7 +490,7 @@ function markRowsFailed(rows: OutboxRow[], now: string, error: string): void {
function markEntityFailed(entityType: string, entityId: string, error: string): void {
const table = entityTable(entityType);
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db || !table) return;
db.prepare(`
UPDATE ${table}
@@ -495,7 +500,7 @@ function markEntityFailed(entityType: string, entityId: string, error: string):
}
function countOutbox(status: string): number {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return 0;
const row = db
.prepare("SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?")
@@ -504,7 +509,7 @@ function countOutbox(status: string): number {
}
function readFailureSummary(): DigitalEmployeeSyncFailureSummary {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return { total: 0, dueForRetry: 0, backoff: 0, byEntityType: [] };
const rows = db
.prepare(
@@ -544,7 +549,7 @@ function readRecentFailures(
config: DigitalEmployeeSyncConfig,
limit = 3,
): DigitalEmployeeSyncFailure[] {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db
.prepare(
@@ -593,7 +598,7 @@ function readAttemptHistory(
outboxId: string,
limit = 5,
): DigitalEmployeeSyncAttempt[] {
const db = getDb();
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db
.prepare(