1+import fs from "node:fs/promises";
2+import path from "node:path";
3+import { tryCronScheduleIdentity } from "../cron/schedule-identity.js";
4+import type {
5+CronConfigJobRuntimeEntry,
6+LoadedCronStore,
7+QuarantinedCronConfigJob,
8+} from "../cron/store.js";
9+import type { CronStoreFile } from "../cron/types.js";
10+import { isRecord } from "../shared/record-coerce.js";
11+import { normalizeOptionalString } from "../shared/string-coerce.js";
12+import { parseJsonWithJson5Fallback } from "../utils/parse-json-compat.js";
13+14+const LEGACY_CRON_ARCHIVE_SUFFIX = ".migrated";
15+16+function resolveLegacyCronStatePath(storePath: string): string {
17+if (storePath.endsWith(".json")) {
18+return storePath.replace(/\.json$/, "-state.json");
19+}
20+return `${storePath}-state.json`;
21+}
22+23+async function legacyCronFileExists(filePath: string): Promise<boolean> {
24+return fs
25+.access(filePath)
26+.then(() => true)
27+.catch(() => false);
28+}
29+30+async function archiveLegacyCronFile(filePath: string): Promise<void> {
31+if (!(await legacyCronFileExists(filePath))) {
32+return;
33+}
34+const archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}`;
35+if (await legacyCronFileExists(archivePath)) {
36+return;
37+}
38+await fs.rename(filePath, archivePath).catch(() => undefined);
39+}
40+41+function parseCronStateFile(raw: string): {
42+version: 1;
43+jobs: Record<string, CronConfigJobRuntimeEntry>;
44+} | null {
45+try {
46+const parsed = parseJsonWithJson5Fallback(raw);
47+if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
48+return null;
49+}
50+const record = parsed as Record<string, unknown>;
51+if (
52+record.version !== 1 ||
53+typeof record.jobs !== "object" ||
54+record.jobs === null ||
55+Array.isArray(record.jobs)
56+) {
57+return null;
58+}
59+return {
60+version: 1,
61+jobs: record.jobs as Record<string, CronConfigJobRuntimeEntry>,
62+};
63+} catch {
64+return null;
65+}
66+}
67+68+function getRawCronJobs(parsed: unknown): unknown[] {
69+return Array.isArray(parsed)
70+ ? parsed
71+ : isRecord(parsed) && Array.isArray(parsed.jobs)
72+ ? parsed.jobs
73+ : [];
74+}
75+76+function cloneConfigJobs(jobs: Array<Record<string, unknown>>): Array<Record<string, unknown>> {
77+return jobs.map((job) => structuredClone(job));
78+}
79+80+async function loadStateFile(
81+statePath: string,
82+): Promise<{ version: 1; jobs: Record<string, CronConfigJobRuntimeEntry> } | null> {
83+let raw: string;
84+try {
85+raw = await fs.readFile(statePath, "utf-8");
86+} catch (err) {
87+if ((err as { code?: unknown })?.code === "ENOENT") {
88+return null;
89+}
90+throw new Error(`Failed to read cron state at ${statePath}: ${String(err)}`, {
91+cause: err,
92+});
93+}
94+95+return parseCronStateFile(raw);
96+}
97+98+function hasInlineState(jobs: Array<Record<string, unknown> | null | undefined>): boolean {
99+return jobs.some(
100+(job) => job != null && isRecord(job.state) && Object.keys(job.state).length > 0,
101+);
102+}
103+104+function ensureJobStateObject(job: CronStoreFile["jobs"][number]): void {
105+if (!isRecord(job.state)) {
106+job.state = {} as never;
107+}
108+}
109+110+function backfillMissingRuntimeFields(job: CronStoreFile["jobs"][number]): void {
111+ensureJobStateObject(job);
112+if (typeof job.updatedAtMs !== "number") {
113+job.updatedAtMs = typeof job.createdAtMs === "number" ? job.createdAtMs : Date.now();
114+}
115+}
116+117+function resolveUpdatedAtMs(job: CronStoreFile["jobs"][number], updatedAtMs: unknown): number {
118+if (typeof updatedAtMs === "number" && Number.isFinite(updatedAtMs)) {
119+return updatedAtMs;
120+}
121+if (typeof job.updatedAtMs === "number" && Number.isFinite(job.updatedAtMs)) {
122+return job.updatedAtMs;
123+}
124+return typeof job.createdAtMs === "number" && Number.isFinite(job.createdAtMs)
125+ ? job.createdAtMs
126+ : Date.now();
127+}
128+129+function mergeStateFileEntry(job: CronStoreFile["jobs"][number], entry: unknown): void {
130+if (!isRecord(entry)) {
131+backfillMissingRuntimeFields(job);
132+return;
133+}
134+job.updatedAtMs = resolveUpdatedAtMs(job, entry.updatedAtMs);
135+job.state = isRecord(entry.state) ? (entry.state as never) : ({} as never);
136+if (
137+typeof entry.scheduleIdentity === "string" &&
138+entry.scheduleIdentity !== tryCronScheduleIdentity(job as unknown as Record<string, unknown>)
139+) {
140+ensureJobStateObject(job);
141+job.state.nextRunAtMs = undefined;
142+}
143+}
144+145+function resolveCronStateId(job: Record<string, unknown>): string | undefined {
146+return normalizeOptionalString(job.id) ?? normalizeOptionalString(job.jobId);
147+}
148+149+export async function legacyCronStoreFilesExist(storePath: string): Promise<boolean> {
150+const resolvedStorePath = path.resolve(storePath);
151+return (
152+(await legacyCronFileExists(resolvedStorePath)) ||
153+(await legacyCronFileExists(resolveLegacyCronStatePath(resolvedStorePath)))
154+);
155+}
156+157+export async function archiveLegacyCronStoreForMigration(storePath: string): Promise<void> {
158+const resolvedStorePath = path.resolve(storePath);
159+await Promise.all([
160+archiveLegacyCronFile(resolvedStorePath),
161+archiveLegacyCronFile(resolveLegacyCronStatePath(resolvedStorePath)),
162+]);
163+}
164+165+export async function loadLegacyCronStoreForMigration(storePath: string): Promise<LoadedCronStore> {
166+const resolvedStorePath = path.resolve(storePath);
167+try {
168+const raw = await fs.readFile(resolvedStorePath, "utf-8");
169+let parsed: unknown;
170+try {
171+parsed = parseJsonWithJson5Fallback(raw);
172+} catch (err) {
173+throw new Error(`Failed to parse cron store at ${resolvedStorePath}: ${String(err)}`, {
174+cause: err,
175+});
176+}
177+const rawJobs = getRawCronJobs(parsed);
178+const configJobIndexes: number[] = [];
179+const configRows: Array<Record<string, unknown>> = [];
180+const configJobRuntimeEntries: CronConfigJobRuntimeEntry[] = [];
181+const invalidConfigRows: QuarantinedCronConfigJob[] = [];
182+for (const [index, row] of rawJobs.entries()) {
183+if (isRecord(row)) {
184+configJobIndexes.push(index);
185+configRows.push(row);
186+} else {
187+invalidConfigRows.push({
188+sourceIndex: index,
189+reason: "non-object-row",
190+raw: structuredClone(row),
191+});
192+}
193+}
194+const store: CronStoreFile = {
195+version: 1,
196+jobs: configRows as never as CronStoreFile["jobs"],
197+};
198+const jobs = store.jobs as unknown as Array<Record<string, unknown>>;
199+const configJobs = cloneConfigJobs(configRows);
200+201+const stateFile = await loadStateFile(resolveLegacyCronStatePath(resolvedStorePath));
202+const hasLegacyInlineState = !stateFile && hasInlineState(jobs);
203+204+if (stateFile) {
205+for (const job of store.jobs) {
206+const stateId = resolveCronStateId(job as unknown as Record<string, unknown>);
207+const entry = stateId ? stateFile.jobs[stateId] : undefined;
208+configJobRuntimeEntries.push(isRecord(entry) ? structuredClone(entry) : {});
209+if (entry) {
210+mergeStateFileEntry(job, entry);
211+} else {
212+backfillMissingRuntimeFields(job);
213+}
214+}
215+} else if (!hasLegacyInlineState) {
216+for (const job of store.jobs) {
217+backfillMissingRuntimeFields(job);
218+}
219+}
220+221+for (const job of store.jobs) {
222+ensureJobStateObject(job);
223+}
224+225+return { store, configJobs, configJobIndexes, configJobRuntimeEntries, invalidConfigRows };
226+} catch (err) {
227+if ((err as { code?: unknown })?.code === "ENOENT") {
228+return {
229+store: { version: 1, jobs: [] },
230+configJobs: [],
231+configJobIndexes: [],
232+configJobRuntimeEntries: [],
233+invalidConfigRows: [],
234+};
235+}
236+throw err;
237+}
238+}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。