File size: 3,050 Bytes
5cb63c1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | import type { DatabaseSync } from "node:sqlite";
import { asRecord } from "@openclaw/normalization-core/record-coerce";
import {
deleteCronJobRowInDatabase,
fingerprintCronJobRows,
loadedCronStoreFromRows,
loadCronRows,
} from "./row-codec.js";
import {
loadCronRuntimeAuthorities,
repairCronRuntimeAuthorityRows,
} from "./runtime-authority-store.js";
import { tryParseJsonObject } from "./scalar-codec.js";
import type { CronJobReadRow } from "./schema.js";
import type { LoadedCronStore } from "./types.js";
type CronLoadWriter = {
write<T>(operation: (db: DatabaseSync) => T, operationLabel: string): T;
committed(): void;
};
function isRetiredCollectionReview(row: CronJobReadRow): boolean {
return (
row.payload_kind === "skillCollectionReview" ||
asRecord(tryParseJsonObject(row.job_json)?.payload).kind === "skillCollectionReview"
);
}
export function loadCronStoreFromDatabase(
database: DatabaseSync,
storeKey: string,
writer?: CronLoadWriter,
): LoadedCronStore {
let rows = loadCronRows(database, storeKey);
const retiredIds = new Set(rows.filter(isRetiredCollectionReview).map((row) => row.job_id));
if (!writer) {
// Hide retired jobs before validation; the next mutable load owns durable deletion.
rows = rows.filter((row) => !retiredIds.has(row.job_id));
} else if (retiredIds.size > 0) {
// Retire generated jobs before runtime validation, including databases already
// on v16. Gateway convergence recreates them with the isolated agent-turn target.
const removed = writer.write((db) => {
const current = loadCronRows(db, storeKey, retiredIds).filter(isRetiredCollectionReview);
for (const row of current) {
deleteCronJobRowInDatabase(db, storeKey, row.job_id);
}
return current.length;
}, "cron.retire-collection-review");
if (removed > 0) {
writer.committed();
}
rows = loadCronRows(database, storeKey);
}
const loaded = loadedCronStoreFromRows(rows);
if (rows.length > 0) {
const authority = loadCronRuntimeAuthorities({
db: database,
storeKey,
jobs: loaded.store.jobs,
});
if (writer) {
repairLoadedCronRuntimeAuthority(writer, {
storeKey,
jobIds: authority.repairJobIds,
});
}
}
return !writer ? loaded : { ...loaded, jobsFingerprint: fingerprintCronJobRows(rows) };
}
function repairLoadedCronRuntimeAuthority(
writer: CronLoadWriter,
params: {
storeKey: string;
jobIds: readonly string[];
},
): void {
if (params.jobIds.length === 0) {
return;
}
const repaired = writer.write((db) => {
const rows = loadCronRows(db, params.storeKey, new Set(params.jobIds));
if (rows.length === 0) {
return false;
}
const loaded = loadedCronStoreFromRows(rows);
return repairCronRuntimeAuthorityRows({
db,
storeKey: params.storeKey,
jobs: loaded.store.jobs,
jobIds: params.jobIds,
});
}, "cron.runtime-authority-repair");
if (repaired) {
writer.committed();
}
}
|