fix(admin): record the pulse rebuild and let a stopped one run again

The rebuild after apply now writes its state beside the dry run
(queued, running, done with counts, failed); a run left queued or
running without a job reads as failed. The card shows it and offers a
new rebuild after a failure, which the route accepts once the method is
applied, without a new dry run.
This commit is contained in:
Marc-André Bombeck committed 2026-10-16 05:55:31 +02:00
1 parent cc9d523bc8
commit f360d221e1
16 files changed
+483 -49

No files matched your search

+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "Ein Probelauf oder eine Neuberechnung läuft noch. Versuche es erneut, sobald er fertig ist.",
"dryRunRequired": "Starte zuerst einen Probelauf. Zum Anwenden braucht es einen, der in den letzten sieben Tagen abgeschlossen wurde.",
"generic": "Die Anfrage ist nicht durchgegangen."
}
},
"rebuild": {
"queued": "Die Neuberechnung wartet auf den Hintergrunddienst.",
"running": "Die Pulstage werden neu berechnet.",
"done": "Pulstage am {at} für {accounts} Konten neu berechnet.",
"failed": "Die Neuberechnung ist abgebrochen. Das App-Log nennt die Ursache; starte sie erneut."
},
"rebuildAgain": "Erneut berechnen",
"rebuildStarted": "Neuberechnung gestartet."
}
},
"carrier": "Netzbetreiber",
+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "A dry run or recalculation is still running. Try again once it has finished.",
"dryRunRequired": "Run a dry run first. Applying needs one that finished in the last seven days.",
"generic": "The request did not go through."
}
},
"rebuild": {
"queued": "The recalculation is waiting for the background worker.",
"running": "The pulse days are being recalculated.",
"done": "Pulse days recalculated on {at} for {accounts} accounts.",
"failed": "The recalculation stopped. The app log names the cause; run it again."
},
"rebuildAgain": "Recalculate again",
"rebuildStarted": "Recalculation started."
}
},
"carrier": "Carrier",
+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "Todavía hay una prueba o un recálculo en marcha. Inténtalo de nuevo cuando termine.",
"dryRunRequired": "Ejecuta primero una prueba. Para aplicar hace falta una terminada en los últimos siete días.",
"generic": "La solicitud no se completó."
}
},
"rebuild": {
"queued": "El recálculo espera al proceso en segundo plano.",
"running": "Se están recalculando los días de pulso.",
"done": "Días de pulso recalculados el {at} para {accounts} cuentas.",
"failed": "El recálculo se detuvo. El registro de la app indica la causa; vuelve a ejecutarlo."
},
"rebuildAgain": "Recalcular de nuevo",
"rebuildStarted": "Recálculo iniciado."
}
},
"carrier": "Operador",
+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "Une simulation ou un recalcul est encore en cours. Réessayez une fois terminé.",
"dryRunRequired": "Lancez d'abord une simulation. L'application en exige une terminée au cours des sept derniers jours.",
"generic": "La requête n'a pas abouti."
}
},
"rebuild": {
"queued": "Le recalcul attend le processus en arrière-plan.",
"running": "Les jours de pouls sont en cours de recalcul.",
"done": "Jours de pouls recalculés le {at} pour {accounts} comptes.",
"failed": "Le recalcul s’est arrêté. Le journal de l’app en indique la cause ; relancez-le."
},
"rebuildAgain": "Recalculer à nouveau",
"rebuildStarted": "Recalcul lancé."
}
},
"carrier": "Opérateur",
+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "Una prova o un ricalcolo è ancora in corso. Riprova quando è terminato.",
"dryRunRequired": "Avvia prima una prova. Per applicare ne serve una terminata negli ultimi sette giorni.",
"generic": "La richiesta non è andata a buon fine."
}
},
"rebuild": {
"queued": "Il ricalcolo attende il processo in background.",
"running": "I giorni del polso sono in fase di ricalcolo.",
"done": "Giorni del polso ricalcolati il {at} per {accounts} account.",
"failed": "Il ricalcolo si è interrotto. Il log dell’app indica la causa; avvialo di nuovo."
},
"rebuildAgain": "Ricalcola di nuovo",
"rebuildStarted": "Ricalcolo avviato."
}
},
"carrier": "Operatore",
+9 -1
View File
@@ -8198,7 +8198,15 @@
"busy": "시험 실행이나 재계산이 아직 진행 중입니다. 끝난 뒤 다시 시도하세요.",
"dryRunRequired": "먼저 시험 실행을 하세요. 적용하려면 최근 7일 안에 끝난 시험 실행이 필요합니다.",
"generic": "요청이 처리되지 않았습니다."
}
},
"rebuild": {
"queued": "재계산이 백그라운드 작업을 기다리고 있습니다.",
"running": "맥박 일별 값을 재계산하고 있습니다.",
"done": "{at}에 계정 {accounts}개의 맥박 일별 값을 재계산했습니다.",
"failed": "재계산이 중단되었습니다. 앱 로그에 원인이 있습니다. 다시 실행하세요."
},
"rebuildAgain": "다시 계산",
"rebuildStarted": "재계산을 시작했습니다."
}
},
"carrier": "통신사",
+9 -1
View File
@@ -8223,7 +8223,15 @@
"busy": "Próba lub przeliczanie wciąż trwa. Spróbuj ponownie, gdy się zakończy.",
"dryRunRequired": "Najpierw uruchom próbę. Do zastosowania potrzebna jest próba zakończona w ciągu ostatnich siedmiu dni.",
"generic": "Żądanie nie zostało zrealizowane."
}
},
"rebuild": {
"queued": "Ponowne przeliczenie czeka na proces w tle.",
"running": "Dni pulsu są przeliczane.",
"done": "Dni pulsu przeliczone {at} dla kont: {accounts}.",
"failed": "Przeliczanie zostało przerwane. Przyczynę podaje log aplikacji; uruchom je ponownie."
},
"rebuildAgain": "Przelicz ponownie",
"rebuildStarted": "Przeliczanie rozpoczęte."
}
},
"carrier": "Operator",
@@ -66,10 +66,14 @@ const post = (body: unknown) =>
const NOW = new Date("2026-10-16T12:00:00Z");
const send = vi.fn();
function settings(dryRun: unknown, method = "hour_mean_v1") {
function settings(
dryRun: unknown,
method = "hour_mean_v1",
appliedAt: Date | null = null,
) {
vi.mocked(prisma.appSettings.findUnique).mockResolvedValue({
pulseDayMeanMethod: method,
pulseDayMeanAppliedAt: null,
pulseDayMeanAppliedAt: appliedAt,
pulseDayMeanDryRunJson: dryRun,
} as never);
}
@@ -136,6 +140,7 @@ describe("GET", () => {
method: "hour_mean_v1",
appliedAt: null,
lastDryRun: doneAt("2026-10-15T08:00:00.000Z"),
lastRebuild: null,
});
});
@@ -199,7 +204,9 @@ describe("POST dry_run", () => {
const res = await post({ action: "dry_run" });
expect(res.status).toBe(409);
expect(vi.mocked(prisma.appSettings.upsert).mock.lastCall?.[0]).toEqual(
expect.objectContaining({ update: { pulseDayMeanDryRunJson: previous } }),
expect.objectContaining({
update: { pulseDayMeanDryRunJson: { ...previous, rebuild: null } },
}),
);
});
});
@@ -264,10 +271,71 @@ describe("POST apply", () => {
});
});
describe("POST rebuild", () => {
const failedRebuild = {
status: "failed",
startedAt: "2026-10-15T09:00:00.000Z",
finishedAt: "2026-10-15T09:05:00.000Z",
accounts: null,
rows: null,
};
it("is refused before any apply", async () => {
settings(doneAt("2026-10-15T08:00:00.000Z"));
const res = await post({ action: "rebuild" });
expect(res.status).toBe(409);
expect((await res.json()).meta.errorCode).toBe(
"maintenance.pulseDayMean.dryRunRequired",
);
expect(send).not.toHaveBeenCalled();
});
it("runs the recalculation again after a stopped one, without a new dry run", async () => {
// The dry run is three weeks old: apply would refuse it, a rebuild does not.
settings(
{ ...doneAt("2026-09-20T08:00:00.000Z"), rebuild: failedRebuild },
"weighted_hour_mean_v2",
new Date("2026-09-20T09:00:00.000Z"),
);
const res = await post({ action: "rebuild" });
expect(res.status).toBe(202);
expect(send).toHaveBeenCalledWith(
"pulse-day-mean",
{ action: "rebuild" },
expect.anything(),
);
expect(
vi.mocked(prisma.appSettings.upsert).mock.lastCall?.[0],
).toMatchObject({
update: {
pulseDayMeanDryRunJson: { rebuild: { status: "queued" } },
},
});
expect(auditLog).toHaveBeenCalledWith(
"admin.pulseDayMean.rebuild",
expect.anything(),
);
});
it("shows a rebuild left queued with no job behind it as failed", async () => {
settings(
{
...doneAt("2026-10-15T08:00:00.000Z"),
rebuild: { ...failedRebuild, status: "running", finishedAt: null },
},
"weighted_hour_mean_v2",
new Date("2026-10-15T09:00:00.000Z"),
);
const body = await (await get()).json();
expect(body.data.lastRebuild.status).toBe("failed");
expect(body.data.lastDryRun.status).toBe("done");
});
});
describe("POST body", () => {
it("is strict", async () => {
expect((await post({ action: "apply", force: true })).status).toBe(422);
expect((await post({ action: "rebuild" })).status).toBe(422);
expect((await post({ action: "purge" })).status).toBe(422);
expect(send).not.toHaveBeenCalled();
});
@@ -4,7 +4,7 @@
*
* GET answers the active method, when it was applied and the last dry run
* (status and result: per-account ordinal labels only, never an id). POST
* `{ action: "dry_run" | "apply" }` answers 202 `{ enqueued }`: a dry run
* `{ action: "dry_run" | "apply" | "rebuild" }` answers 202 `{ enqueued }`: a dry run
* computes both methods for every account and day and writes nothing but its
* result; `apply` needs a successful dry run from the last seven days (409
* `maintenance.pulseDayMean.dryRunRequired`), switches the method and
@@ -31,6 +31,7 @@ import {
PULSE_DAY_MEAN_SEND_OPTIONS,
readPulseDayMeanStatus,
writeDryRun,
writeRebuild,
type PulseDayMeanPayload,
} from "@/lib/jobs/pulse-day-mean";
import { annotate } from "@/lib/logging/context";
@@ -41,6 +42,7 @@ import {
import {
hasFreshDryRun,
pulseDayMeanActionSchema,
type PulseDayMeanStatus,
} from "@/lib/validations/pulse-day-mean";
export const dynamic = "force-dynamic";
@@ -54,6 +56,33 @@ function busy() {
});
}
/**
* Queue the rebuild and record it as queued, so the card shows it from the
* moment it is asked for and a rebuild that never starts reads as failed.
* When another run holds the singleton, the previous rebuild state is put
* back and the answer is false.
*/
async function sendRebuild(
boss: NonNullable<ReturnType<typeof getGlobalBoss>>,
previous: PulseDayMeanStatus["lastRebuild"],
): Promise<boolean> {
await writeRebuild({
status: "queued",
startedAt: new Date().toISOString(),
finishedAt: null,
accounts: null,
rows: null,
});
const payload: PulseDayMeanPayload = { action: "rebuild" };
const jobId = await boss.send(
PULSE_DAY_MEAN_QUEUE,
payload,
PULSE_DAY_MEAN_SEND_OPTIONS,
);
if (jobId === null) await writeRebuild(previous);
return jobId !== null;
}
export const GET = apiHandler(async () => {
await requireAdmin();
const status = await readPulseDayMeanStatus();
@@ -110,6 +139,28 @@ export const POST = apiHandler(async (request: NextRequest) => {
return apiSuccess({ enqueued: true }, 202);
}
if (action === "rebuild") {
// The recalculation again, after an apply whose rebuild stopped. The
// method is already switched; the dry run that justified it stands.
const status = await readPulseDayMeanStatus();
if (status.appliedAt === null) {
return apiError(
"Nothing applied yet; run a dry run and apply first",
409,
{ errorCode: "maintenance.pulseDayMean.dryRunRequired" },
);
}
const enqueued = await sendRebuild(boss, status.lastRebuild);
if (!enqueued) return busy();
await auditLog("admin.pulseDayMean.rebuild", {
userId: user.id,
ipAddress: getClientIp(request),
details: { method: status.method },
});
annotate({ meta: { pulse_day_mean_enqueued: true } });
return apiSuccess({ enqueued: true }, 202);
}
if (!hasFreshDryRun(await readPulseDayMeanStatus())) {
return apiError(
"Run a dry run first; apply needs one finished in the last seven days",
@@ -135,13 +186,10 @@ export const POST = apiHandler(async (request: NextRequest) => {
},
});
setActivePulseDayMeanMethod(APPLIED_METHOD);
const payload: PulseDayMeanPayload = { action: "rebuild" };
const jobId = await boss.send(
PULSE_DAY_MEAN_QUEUE,
payload,
PULSE_DAY_MEAN_SEND_OPTIONS,
const enqueued = await sendRebuild(
boss,
(await readPulseDayMeanStatus()).lastRebuild,
);
const enqueued = jobId !== null;
await auditLog("admin.pulseDayMean.apply", {
userId: user.id,
@@ -37,7 +37,12 @@ const RESULT = {
const status = (
lastDryRun: PulseDayMeanStatus["lastDryRun"],
method: PulseDayMeanStatus["method"] = "hour_mean_v1",
): PulseDayMeanStatus => ({ method, appliedAt: null, lastDryRun });
): PulseDayMeanStatus => ({
method,
appliedAt: null,
lastDryRun,
lastRebuild: null,
});
const done = (finishedAt: string) => ({
status: "done" as const,
@@ -113,4 +118,33 @@ describe("PulseDayMeanCard", () => {
expect(de).toContain("Dünne Stunden gewichtet");
expect(de).toContain("Probelauf");
});
it("shows a stopped recalculation and offers it again without a dry run", () => {
const failed: PulseDayMeanStatus = {
...status(done("2026-10-01T10:00:00Z"), "weighted_hour_mean_v2"),
appliedAt: "2026-10-01T11:00:00Z",
lastRebuild: {
status: "failed",
startedAt: "2026-10-01T11:00:00Z",
finishedAt: "2026-10-01T11:05:00Z",
accounts: null,
rows: null,
},
};
const html = render(failed);
expect(html).toContain('data-state="failed"');
expect(html).toContain('data-testid="admin-pulse-day-mean-rebuild"');
const finished = render({
...failed,
lastRebuild: {
...failed.lastRebuild!,
status: "done",
accounts: 3,
rows: 900,
},
});
expect(finished).not.toContain("admin-pulse-day-mean-rebuild");
expect(finished).toContain('data-slot="pulse-day-mean-rebuild"');
});
});
+48 -3
View File
@@ -57,8 +57,13 @@ export function PulseDayMeanCard() {
queryFn: () =>
apiGet<PulseDayMeanStatus>(ENDPOINT, { credentials: "include" }),
refetchInterval: (query) => {
const state = query.state.data?.lastDryRun?.status;
return state === "queued" || state === "running" ? POLL_MS : false;
const states = [
query.state.data?.lastDryRun?.status,
query.state.data?.lastRebuild?.status,
];
return states.some((state) => state === "queued" || state === "running")
? POLL_MS
: false;
},
});
@@ -101,6 +106,8 @@ export function PulseDayMeanCard() {
? t("admin.section.pulseDayMean.applied")
: t("admin.section.pulseDayMean.appliedNoRebuild"),
);
} else if (action === "rebuild") {
toast.success(t("admin.section.pulseDayMean.rebuildStarted"));
} else {
toast.success(t("admin.section.pulseDayMean.dryRunStarted"));
}
@@ -125,9 +132,16 @@ export function PulseDayMeanCard() {
const s = status.data;
const dryRun = s?.lastDryRun ?? null;
const result = dryRun?.status === "done" ? dryRun.result : null;
const active = dryRun?.status === "queued" || dryRun?.status === "running";
const rebuild = s?.lastRebuild ?? null;
const rebuilding =
rebuild?.status === "queued" || rebuild?.status === "running";
const active =
dryRun?.status === "queued" || dryRun?.status === "running" || rebuilding;
const weighted = s?.method === "weighted_hour_mean_v2";
const canApply = !!s && !active && hasFreshDryRun(s);
// A stopped recalculation is run again from here; the method is already
// switched, so it needs no new dry run.
const canRebuild = !!s?.appliedAt && rebuild?.status === "failed";
return (
<SettingsCard data-testid="admin-pulse-day-mean">
@@ -252,6 +266,25 @@ export function PulseDayMeanCard() {
</p>
) : null}
{rebuild ? (
<p
className={
rebuild.status === "failed"
? "text-destructive text-sm"
: "text-muted-foreground text-xs"
}
data-slot="pulse-day-mean-rebuild"
data-state={rebuild.status}
>
{rebuild.status === "done"
? t("admin.section.pulseDayMean.rebuild.done", {
at: fmt.dateTime(rebuild.finishedAt ?? rebuild.startedAt),
accounts: fmt.number(rebuild.accounts ?? 0),
})
: t(`admin.section.pulseDayMean.rebuild.${rebuild.status}`)}
</p>
) : null}
<SettingsCardActions>
<Button
size="sm"
@@ -264,6 +297,18 @@ export function PulseDayMeanCard() {
<FlaskConical className="size-3.5" aria-hidden />
{t("admin.section.pulseDayMean.dryRun")}
</Button>
{canRebuild ? (
<Button
size="sm"
variant="outline"
className="min-h-11 sm:min-h-9"
disabled={run.isPending}
onClick={() => run.mutate("rebuild")}
data-testid="admin-pulse-day-mean-rebuild"
>
{t("admin.section.pulseDayMean.rebuildAgain")}
</Button>
) : null}
<AlertDialog>
<AlertDialogTrigger asChild>
<Button
@@ -0,0 +1,73 @@
/**
* The pulse rollup rebuild is on the record: a stopped one reads as failed
* beside the dry run, so the admin card can offer it again.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const db = vi.hoisted(() => {
let stored: unknown = {
status: "done",
startedAt: "2026-10-15T08:00:00.000Z",
finishedAt: "2026-10-15T08:10:00.000Z",
result: null,
};
return {
get stored() {
return stored;
},
reset(value: unknown) {
stored = value;
},
appSettings: {
findUnique: vi.fn(async () => ({ pulseDayMeanDryRunJson: stored })),
upsert: vi.fn(
async (args: { update: { pulseDayMeanDryRunJson: unknown } }) => {
stored = args.update.pulseDayMeanDryRunJson;
return {};
},
),
},
$queryRaw: vi.fn(),
};
});
vi.mock("@/lib/db", () => ({
prisma: { appSettings: db.appSettings, $queryRaw: db.$queryRaw },
}));
vi.mock("@/lib/jobs/job-lock", () => ({
withJobLock: async (_key: string, run: () => Promise<unknown>) => ({
ran: true,
result: await run(),
}),
}));
vi.mock("@/lib/logging/signal", () => ({ logCaught: vi.fn() }));
import { handlePulseDayMean } from "../pulse-day-mean";
const job = [{ data: { action: "rebuild" } }] as never;
beforeEach(() => {
db.$queryRaw.mockReset();
});
describe("the pulse rebuild state", () => {
it("records a stopped rebuild as failed, beside the dry run", async () => {
db.$queryRaw.mockRejectedValue(new Error("statement timeout"));
const outcome = await handlePulseDayMean(job);
expect(outcome.ok).toBe(false);
expect(db.stored).toMatchObject({
status: "done",
rebuild: { status: "failed", accounts: null },
});
});
it("records a finished rebuild with what it did", async () => {
// No account holds pulse: done, nothing written.
db.$queryRaw.mockResolvedValue([]);
const outcome = await handlePulseDayMean(job);
expect(outcome.ok).toBe(true);
expect(db.stored).toMatchObject({
rebuild: { status: "done", accounts: 0, rows: 0 },
});
});
});
+92 -20
View File
@@ -41,10 +41,12 @@ import {
} from "@/lib/rollups/measurement-rollups";
import {
PULSE_DAY_MEAN_CHANGE_BUCKETS,
pulseDayMeanDryRunSchema,
pulseDayMeanStoredRunsSchema,
type PulseDayMeanDryRun,
type PulseDayMeanDryRunResult,
type PulseDayMeanRebuild,
type PulseDayMeanStatus,
type PulseDayMeanStoredRuns,
} from "@/lib/validations/pulse-day-mean";
export const PULSE_DAY_MEAN_QUEUE = "pulse-day-mean";
@@ -92,15 +94,26 @@ export async function isPulseDayMeanRunActive(): Promise<boolean> {
}
}
function parseDryRun(value: unknown): PulseDayMeanDryRun | null {
const parsed = pulseDayMeanDryRunSchema.safeParse(value);
function parseStoredRuns(value: unknown): PulseDayMeanStoredRuns | null {
const parsed = pulseDayMeanStoredRunsSchema.safeParse(value);
return parsed.success ? parsed.data : null;
}
/** A run still marked queued or running with no live job behind it failed. */
function settle<T extends { status: PulseDayMeanDryRun["status"] }>(
run: T | null,
active: boolean,
): T | null {
if (run && (run.status === "queued" || run.status === "running") && !active) {
return { ...run, status: "failed" };
}
return run;
}
/**
* The stored method, when it was applied, and the last dry run. A dry run
* still marked queued or running with no live job behind it (the worker went
* away mid-run) reads as failed.
* The stored method, when it was applied, the last dry run and the last
* rebuild. A run still marked queued or running with no live job behind it
* (the worker went away mid-run) reads as failed.
*/
export async function readPulseDayMeanStatus(): Promise<PulseDayMeanStatus> {
const row = await prisma.appSettings.findUnique({
@@ -111,32 +124,59 @@ export async function readPulseDayMeanStatus(): Promise<PulseDayMeanStatus> {
pulseDayMeanDryRunJson: true,
},
});
let lastDryRun = parseDryRun(row?.pulseDayMeanDryRunJson ?? null);
if (
lastDryRun &&
(lastDryRun.status === "queued" || lastDryRun.status === "running") &&
!(await isPulseDayMeanRunActive())
) {
lastDryRun = { ...lastDryRun, status: "failed" };
}
const stored = parseStoredRuns(row?.pulseDayMeanDryRunJson ?? null);
const pending =
stored !== null &&
[stored.status, stored.rebuild?.status].some(
(status) => status === "queued" || status === "running",
);
const active = pending ? await isPulseDayMeanRunActive() : false;
const { rebuild, ...dryRun } = stored ?? { rebuild: null };
return {
method: isPulseDayMeanMethod(row?.pulseDayMeanMethod)
? row.pulseDayMeanMethod
: DEFAULT_PULSE_DAY_MEAN_METHOD,
appliedAt: row?.pulseDayMeanAppliedAt?.toISOString() ?? null,
lastDryRun,
lastDryRun: stored ? settle(dryRun as PulseDayMeanDryRun, active) : null,
lastRebuild: settle(rebuild ?? null, active),
};
}
/** The only write a dry run makes. */
export async function writeDryRun(run: PulseDayMeanDryRun): Promise<void> {
async function readStoredRuns(): Promise<PulseDayMeanStoredRuns | null> {
const row = await prisma.appSettings.findUnique({
where: { id: "singleton" },
select: { pulseDayMeanDryRunJson: true },
});
return parseStoredRuns(row?.pulseDayMeanDryRunJson ?? null);
}
async function writeStoredRuns(runs: PulseDayMeanStoredRuns): Promise<void> {
await prisma.appSettings.upsert({
where: { id: "singleton" },
create: { id: "singleton", pulseDayMeanDryRunJson: run },
update: { pulseDayMeanDryRunJson: run },
create: { id: "singleton", pulseDayMeanDryRunJson: runs },
update: { pulseDayMeanDryRunJson: runs },
});
}
/** The only write a dry run makes. The last rebuild stays as it was. */
export async function writeDryRun(run: PulseDayMeanDryRun): Promise<void> {
const stored = await readStoredRuns();
await writeStoredRuns({ ...run, rebuild: stored?.rebuild ?? null });
}
/**
* Record the rebuild's state beside the dry run it followed. A rebuild only
* runs after an apply, and an apply needs a dry run, so there is always one
* to sit beside; without it there is nothing to record against.
*/
export async function writeRebuild(
rebuild: PulseDayMeanRebuild | null,
): Promise<void> {
const stored = await readStoredRuns();
if (!stored) return;
await writeStoredRuns({ ...stored, rebuild });
}
// ─── Accounts ─────────────────────────────────────────────
interface PulseAccount {
@@ -354,6 +394,38 @@ export async function rebuildPulseRollups(): Promise<{
// ─── Job ──────────────────────────────────────────────────
async function runRebuild(): Promise<{ accounts: number; rows: number }> {
const startedAt = new Date().toISOString();
await writeRebuild({
status: "running",
startedAt,
finishedAt: null,
accounts: null,
rows: null,
});
try {
const done = await rebuildPulseRollups();
await writeRebuild({
status: "done",
startedAt,
finishedAt: new Date().toISOString(),
...done,
});
return done;
} catch (err) {
await writeRebuild({
status: "failed",
startedAt,
finishedAt: new Date().toISOString(),
accounts: null,
rows: null,
}).catch((writeErr) =>
logCaught("pulse_day_mean.rebuild.status_write_failed", writeErr),
);
throw err;
}
}
async function runDryRun(): Promise<PulseDayMeanDryRunResult> {
const startedAt = new Date().toISOString();
await writeDryRun({
@@ -394,7 +466,7 @@ export async function handlePulseDayMean(
try {
const guarded = await withJobLock(PULSE_DAY_MEAN_QUEUE, async () =>
action === "rebuild"
? { kind: "rebuild" as const, ...(await rebuildPulseRollups()) }
? { kind: "rebuild" as const, ...(await runRebuild()) }
: { kind: "dry_run" as const, result: await runDryRun() },
);
if (!guarded.ran) {
@@ -1,6 +1,6 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
const send = vi.hoisted(() => vi.fn(async () => "job-1"));
const send = vi.hoisted(() => vi.fn(async (..._args: unknown[]) => "job-1"));
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: () => ({ send }),
}));
+33 -4
View File
@@ -9,7 +9,9 @@ import { z } from "zod/v4";
import type { PulseDayMeanMethod } from "@/lib/measurements/day-statistic";
export const pulseDayMeanActionSchema = z.strictObject({
action: z.enum(["dry_run", "apply"]),
// `rebuild` runs the recalculation again after an apply whose rebuild
// stopped, without a new dry run: the method is already switched.
action: z.enum(["dry_run", "apply", "rebuild"]),
});
export type PulseDayMeanActionInput = z.infer<typeof pulseDayMeanActionSchema>;
@@ -46,20 +48,47 @@ const dryRunResultSchema = z.object({
});
export type PulseDayMeanDryRunResult = z.infer<typeof dryRunResultSchema>;
/** `app_settings.pulse_day_mean_dry_run_json`: the last dry run. */
export const pulseDayMeanDryRunSchema = z.object({
status: z.enum(["queued", "running", "done", "failed"]),
const runStatusSchema = z.enum(["queued", "running", "done", "failed"]);
/** The last dry run. */
const pulseDayMeanDryRunSchema = z.object({
status: runStatusSchema,
startedAt: z.string(),
finishedAt: z.string().nullable(),
result: dryRunResultSchema.nullable(),
});
export type PulseDayMeanDryRun = z.infer<typeof pulseDayMeanDryRunSchema>;
/** The last recalculation of the pulse rollups after an apply. */
const pulseDayMeanRebuildSchema = z.object({
status: runStatusSchema,
startedAt: z.string(),
finishedAt: z.string().nullable(),
/** Accounts and rollup rows done; null until the run has finished. */
accounts: z.number().int().nonnegative().nullable(),
rows: z.number().int().nonnegative().nullable(),
});
export type PulseDayMeanRebuild = z.infer<typeof pulseDayMeanRebuildSchema>;
/**
* `app_settings.pulse_day_mean_dry_run_json`: the last dry run, and beside it
* the last rebuild. One column holds both, because the rebuild only ever
* follows an apply that needed a dry run; a blob written before the rebuild
* was recorded has no `rebuild` key and reads as none.
*/
export const pulseDayMeanStoredRunsSchema = pulseDayMeanDryRunSchema.extend({
rebuild: pulseDayMeanRebuildSchema.nullable().optional(),
});
export type PulseDayMeanStoredRuns = z.infer<
typeof pulseDayMeanStoredRunsSchema
>;
/** `GET /api/admin/maintenance/pulse-day-mean`. */
export interface PulseDayMeanStatus {
method: PulseDayMeanMethod;
appliedAt: string | null;
lastDryRun: PulseDayMeanDryRun | null;
lastRebuild: PulseDayMeanRebuild | null;
}
/** How old a finished dry run may be for the apply step to accept it. */
@@ -38,7 +38,7 @@ import { readLiveBuckets } from "@/lib/measurements/daily-series-read";
import { readCanonicalRollupBuckets } from "@/lib/rollups/measurement-read";
import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
import { invalidateUserTimezone } from "@/lib/tz/resolver";
import { pulseDayMeanDryRunSchema } from "@/lib/validations/pulse-day-mean";
import { pulseDayMeanStoredRunsSchema } from "@/lib/validations/pulse-day-mean";
import type { Job } from "pg-boss";
const prisma = getPrismaClient();
@@ -276,7 +276,9 @@ describe("the dry run", () => {
const stored = await prisma.appSettings.findUniqueOrThrow({
where: { id: "singleton" },
});
const run = pulseDayMeanDryRunSchema.parse(stored.pulseDayMeanDryRunJson);
const run = pulseDayMeanStoredRunsSchema.parse(
stored.pulseDayMeanDryRunJson,
);
expect(run.status).toBe("done");
expect(run.finishedAt).not.toBeNull();
expect(run.result).toEqual({
@@ -324,6 +326,13 @@ describe("the rebuild after apply", () => {
});
const outcome = await handlePulseDayMean(job("rebuild"));
expect(outcome.ok).toBe(true);
// The rebuild is on the record beside the dry run, with what it did.
const { readPulseDayMeanStatus } =
await import("@/lib/jobs/pulse-day-mean");
const status = await readPulseDayMeanStatus();
expect(status.lastRebuild).toMatchObject({ status: "done" });
expect(status.lastRebuild?.accounts).toBeGreaterThan(0);
expect(status.lastDryRun?.status).toBe("done");
expect(await dayMean(A, D1)).toBeCloseTo(V2, 6);
expect(await dayMean(A, D2)).toBeCloseTo(70, 6);