130 lines
4.1 KiB
TypeScript
130 lines
4.1 KiB
TypeScript
import type { FastifyBaseLogger } from 'fastify';
|
|
import { query, queryOne } from './db.js';
|
|
|
|
/**
|
|
* Long-running jobs that must not sit in front of an HTTP request: the NDA
|
|
* refresh walk and the signature sync. Both talk to a rate-limited API for
|
|
* minutes at a time, so the route starts them and returns, and the UI polls
|
|
* the state out of app_meta.
|
|
*
|
|
* At most one run of each at a time, claimed atomically so two clicks on the
|
|
* same button cannot start two jobs.
|
|
*/
|
|
|
|
export interface TaskKeys {
|
|
/** idle | running | error:<message> */
|
|
state: string;
|
|
/**
|
|
* ISO timestamp of the last completed run — written by the **job**, never
|
|
* by this module. Each job owns its own key and the two sets are disjoint
|
|
* (`ds_*_refresh_*` vs `ds_*_sync_*`), because `ds_last_refresh_at` is not
|
|
* merely a display value: it is the boundary the sync uses to decide what
|
|
* the refresh can no longer reach. See startTask().
|
|
*/
|
|
lastAt: string;
|
|
/** JSON result of the last completed run. */
|
|
lastResult: string;
|
|
}
|
|
|
|
export async function getMeta(key: string): Promise<string | null> {
|
|
const row = await queryOne<{ value: string | null }>(
|
|
'SELECT value FROM app_meta WHERE key = $1',
|
|
[key],
|
|
);
|
|
return row?.value ?? null;
|
|
}
|
|
|
|
export async function setMeta(key: string, value: string): Promise<void> {
|
|
await query(
|
|
`INSERT INTO app_meta (key, value) VALUES ($1, $2)
|
|
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`,
|
|
[key, value],
|
|
);
|
|
}
|
|
|
|
/**
|
|
* The conditional upsert *is* the lock: it returns nothing when somebody else
|
|
* already holds the slot.
|
|
*/
|
|
async function claim(keys: TaskKeys): Promise<boolean> {
|
|
const claimed = await queryOne<{ key: string }>(
|
|
`INSERT INTO app_meta (key, value) VALUES ($1, 'running')
|
|
ON CONFLICT (key) DO UPDATE SET value = 'running', updated_at = now()
|
|
WHERE app_meta.value <> 'running'
|
|
RETURNING key`,
|
|
[keys.state],
|
|
);
|
|
return Boolean(claimed);
|
|
}
|
|
|
|
/**
|
|
* Tasks live in this process, so a "running" found at startup can only be the
|
|
* remains of a killed one.
|
|
*/
|
|
export async function clearStaleTask(
|
|
keys: TaskKeys,
|
|
label: string,
|
|
log: FastifyBaseLogger,
|
|
): Promise<void> {
|
|
const stale = await queryOne<{ key: string }>(
|
|
`UPDATE app_meta SET value = 'idle', updated_at = now()
|
|
WHERE key = $1 AND value = 'running' RETURNING key`,
|
|
[keys.state],
|
|
);
|
|
if (stale) log.warn(`[${label}] found a stale "running" state at startup, reset to idle`);
|
|
}
|
|
|
|
/**
|
|
* Claims the slot and runs `job` detached. `false` means somebody else is
|
|
* already running it — the caller answers 409.
|
|
*
|
|
* It deliberately does **not** stamp `keys.lastAt`. A shared helper writing
|
|
* "this ran at" on every resolved job is how `ds_last_refresh_at` came to move
|
|
* forward on runs that had walked nothing: the refresh cursor then pointed
|
|
* past requests that were never mirrored, and nothing downstream could tell
|
|
* the difference. Each job writes its own `lastAt` at the point where it knows
|
|
* its work is done and stored.
|
|
*/
|
|
export async function startTask<T>(
|
|
keys: TaskKeys,
|
|
label: string,
|
|
log: FastifyBaseLogger,
|
|
job: () => Promise<T>,
|
|
): Promise<boolean> {
|
|
if (!(await claim(keys))) return false;
|
|
|
|
void (async () => {
|
|
try {
|
|
const result = await job();
|
|
await setMeta(keys.lastResult, JSON.stringify(result));
|
|
await setMeta(keys.state, 'idle');
|
|
} catch (err) {
|
|
const message = (err as Error).message;
|
|
log.error(`[${label}] failed: ${message}`);
|
|
// Kept in the state so the UI can show why, rather than a silent idle.
|
|
await setMeta(keys.state, `error:${message}`.slice(0, 400)).catch(() => {});
|
|
}
|
|
})();
|
|
|
|
return true;
|
|
}
|
|
|
|
export interface TaskState {
|
|
state: string;
|
|
last_at: string | null;
|
|
last_result: unknown;
|
|
}
|
|
|
|
export async function readTaskState(keys: TaskKeys): Promise<TaskState> {
|
|
const [state, lastAt, lastResult] = await Promise.all([
|
|
getMeta(keys.state),
|
|
getMeta(keys.lastAt),
|
|
getMeta(keys.lastResult),
|
|
]);
|
|
return {
|
|
state: state ?? 'idle',
|
|
last_at: lastAt,
|
|
last_result: lastResult ? (JSON.parse(lastResult) as unknown) : null,
|
|
};
|
|
}
|