mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
Merge branch 'v1422/fixes' into release/v1.42.2
This commit is contained in:
commit
3c4e053dcd
83 files changed
+1808
-258
No files matched your search
@@ -0,0 +1,97 @@
|
||||
/**
|
||||
* Raw SQL never reads the clock through the session time zone.
|
||||
*
|
||||
* Stored times are UTC in `timestamp` columns without a zone. Postgres
|
||||
* compares such a column with `NOW()` / `CURRENT_TIMESTAMP` /
|
||||
* `clock_timestamp()` by converting through the session time zone, and
|
||||
* assigns one of them to such a column the same way, so on a database whose
|
||||
* default is not UTC a fresh passkey challenge read as already expired. Every
|
||||
* connection is pinned to UTC (`@/lib/db-session`), and on top of that every
|
||||
* raw-SQL clock read writes `NOW() AT TIME ZONE 'UTC'`, which yields the naive
|
||||
* UTC timestamp whatever the session says.
|
||||
*
|
||||
* The few bare reads left are listed below with the reason they are safe:
|
||||
* each feeds a `timestamptz` column or returns the instant to JavaScript.
|
||||
*/
|
||||
import { readdirSync, readFileSync, statSync } from "node:fs";
|
||||
import { join, relative } from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
const ROOT = join(__dirname, "..", "..");
|
||||
const SRC = join(ROOT, "src");
|
||||
|
||||
/** `file: reason` for every bare clock read that is deliberately left. */
|
||||
const ALLOWED: Record<string, string> = {
|
||||
"src/lib/rollups/mood-rollups.ts": "computed_at is timestamptz",
|
||||
"src/lib/rollups/medication-compliance-rollups.ts":
|
||||
"computed_at is timestamptz",
|
||||
"src/lib/jobs/job-failures.ts": "pg-boss tables are timestamptz",
|
||||
"src/lib/jobs/reminder/registrar-shared.ts": "pg-boss tables are timestamptz",
|
||||
"src/lib/db-compat.ts": "DDL default on a timestamptz column",
|
||||
};
|
||||
|
||||
const SQL_CLOCK =
|
||||
/(?<![.\w])(?:NOW\(\)|CURRENT_TIMESTAMP\b|clock_timestamp\(\))|(?:(?<!=)[<>]|=)\s*now\(\)\s*(?:$|[,)]|[-+]\s*(?:make_interval|INTERVAL))|\bnow\(\)\s*[-+]\s*(?:make_interval|INTERVAL)/;
|
||||
const ANCHORED =
|
||||
/(?:NOW\(\)|CURRENT_TIMESTAMP|clock_timestamp\(\)) AT TIME ZONE 'UTC'/g;
|
||||
|
||||
function walk(dir: string, out: string[] = []): string[] {
|
||||
for (const name of readdirSync(dir)) {
|
||||
const full = join(dir, name);
|
||||
if (name === "generated" || name === "__tests__") continue;
|
||||
if (statSync(full).isDirectory()) walk(full, out);
|
||||
else if (/\.tsx?$/.test(name) && !/\.test\.tsx?$/.test(name))
|
||||
out.push(full);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function offenders(): string[] {
|
||||
const found: string[] = [];
|
||||
for (const file of walk(SRC)) {
|
||||
const rel = relative(ROOT, file);
|
||||
if (ALLOWED[rel]) continue;
|
||||
readFileSync(file, "utf8")
|
||||
.split("\n")
|
||||
.forEach((line, index) => {
|
||||
if (/^\s*(\/\/|\*|\/\*|--)/.test(line)) return;
|
||||
if (SQL_CLOCK.test(line.replace(ANCHORED, ""))) {
|
||||
found.push(`${rel}:${index + 1}: ${line.trim()}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
describe("raw SQL clock reads are anchored to UTC", () => {
|
||||
it("finds the anchored form across the swept files", () => {
|
||||
// The matcher must see something, or a broken walk would pass.
|
||||
const passkey = readFileSync(join(SRC, "lib/auth/passkey.ts"), "utf8");
|
||||
expect(passkey.match(ANCHORED)?.length).toBeGreaterThanOrEqual(3);
|
||||
expect(SQL_CLOCK.test("AND expires_at > NOW()")).toBe(true);
|
||||
expect(SQL_CLOCK.test("updated_at = now()")).toBe(true);
|
||||
expect(SQL_CLOCK.test("const started = now();")).toBe(false);
|
||||
expect(SQL_CLOCK.test("elapsedMs: () => now() - started,")).toBe(false);
|
||||
expect(SQL_CLOCK.test("const ends = now() + perCallMs;")).toBe(false);
|
||||
expect(
|
||||
SQL_CLOCK.test("deleted_at >= now() - make_interval(days => 1)"),
|
||||
).toBe(true);
|
||||
expect(SQL_CLOCK.test("Date.now()")).toBe(false);
|
||||
});
|
||||
|
||||
it("leaves no bare NOW() / CURRENT_TIMESTAMP outside the allowlist", () => {
|
||||
expect(offenders()).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps the allowlist honest: every entry still has a bare read", () => {
|
||||
for (const rel of Object.keys(ALLOWED)) {
|
||||
const lines = readFileSync(join(ROOT, rel), "utf8")
|
||||
.replace(ANCHORED, "")
|
||||
.split("\n");
|
||||
expect(
|
||||
lines.some((line) => SQL_CLOCK.test(line)),
|
||||
rel,
|
||||
).toBe(true);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -73,11 +73,13 @@ interface ProviderHealthSummary {
|
||||
const CENTRAL_TYPES = [...OPERATOR_HELD_PROVIDER_TYPES];
|
||||
|
||||
export const GET = apiHandler(async () => {
|
||||
await requireAdmin();
|
||||
const { user } = await requireAdmin();
|
||||
annotate({ action: { name: "admin.provider-health.get" } });
|
||||
|
||||
// The UTC day the ledger buckets by — the same key every reservation writes.
|
||||
const dateKey = buildDateKey();
|
||||
// The ledger buckets each account by its own local day; the operator's
|
||||
// "today" is the day in the operator's own zone, which on a single-zone
|
||||
// instance is everyone's.
|
||||
const dateKey = await buildDateKey(user.id);
|
||||
|
||||
const groups = await prisma.providerHealth.groupBy({
|
||||
// `lastStatus` joins the grouping key rather than an aggregate: the
|
||||
|
||||
@@ -145,7 +145,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
// `admin-key` is the operator's own credential and draws the operator
|
||||
// ceiling; an override names the caller's own key, and every other resolved
|
||||
// type is the user's own egress, so those draw the user-plan ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(user.id);
|
||||
const reservation = await reserveBudget(
|
||||
user.id,
|
||||
AI_TEST_MAX_TOKENS,
|
||||
|
||||
@@ -183,7 +183,7 @@ async function reserveFor(
|
||||
providerType: ProviderChainType,
|
||||
tokens: number,
|
||||
): Promise<AiRunBudget | null> {
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
tokens,
|
||||
|
||||
@@ -200,7 +200,7 @@ async function handleVisionIndex(
|
||||
const rl = await checkDocumentAiRateLimit(userId);
|
||||
if (!rl.allowed) return documentAiRateLimited(rl);
|
||||
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
DOCUMENT_INDEX_RESERVE_TOKENS,
|
||||
|
||||
@@ -240,6 +240,10 @@ describe("serving-posture header matrix (Class A inline vs Class B attachment)",
|
||||
["image/heic", "photo.heic"],
|
||||
["application/xml", "export.xml"],
|
||||
["application/json", "export.json"],
|
||||
// Upload refuses markup outright; a row that somehow carries one must
|
||||
// still never render in-origin, where its script would run.
|
||||
["image/svg+xml", "drawing.svg"],
|
||||
["text/html", "page.html"],
|
||||
])(
|
||||
"Class B %s → attachment + octet-stream, never inline",
|
||||
async (mime, name) => {
|
||||
|
||||
@@ -122,7 +122,7 @@ async function reserveFor(
|
||||
userId: string,
|
||||
providerType: ProviderChainType,
|
||||
): Promise<AiRunBudget | null> {
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.documentAssist.maxTokens,
|
||||
|
||||
@@ -216,7 +216,7 @@ async function reserveFor(
|
||||
ask: SummaryRequest,
|
||||
providerType: ProviderChainType,
|
||||
): Promise<AiRunBudget | null> {
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(ask.userId);
|
||||
const reservation = await reserveBudget(
|
||||
ask.userId,
|
||||
budgetFor(ask.mode),
|
||||
|
||||
@@ -157,7 +157,7 @@ async function handleTextExtract(
|
||||
// v1.21.0 (F1) — the operator-cost cap applies only when the picked provider
|
||||
// egresses on the operator's own key; a BYOK / Codex / local pick runs on the
|
||||
// user's own plan and gets the generous user-plan ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
OCR_RESERVE_TOKENS.text,
|
||||
@@ -218,7 +218,7 @@ async function handleVisionExtract(
|
||||
// v1.21.0 (F1) — the operator-cost cap applies only when the picked vision
|
||||
// provider egresses on the operator's own key; a BYOK / Codex pick runs on
|
||||
// the user's own plan and gets the generous user-plan ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
OCR_RESERVE_TOKENS.vision,
|
||||
|
||||
@@ -378,7 +378,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
|
||||
AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type"
|
||||
AND "deleted_at" IS NULL
|
||||
AND "value" >= $3
|
||||
AND "measured_at" >= date_trunc('day', $2::timestamptz)
|
||||
AND "measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC')
|
||||
ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text
|
||||
) c
|
||||
ON c.d = date_trunc('day', m."measured_at")
|
||||
|
||||
@@ -225,7 +225,10 @@ async function handleExtract(request: NextRequest): Promise<Response> {
|
||||
chain.map((entry) => entry.providerType),
|
||||
);
|
||||
|
||||
const today = todayOverride ?? buildDateKey();
|
||||
// The account's own today: "tomorrow" in a description means the
|
||||
// person's tomorrow, and the same key is the budget day below.
|
||||
const accountToday = await buildDateKey(userId);
|
||||
const today = todayOverride ?? accountToday;
|
||||
const { systemPrompt, userPrompt } = buildMedicationExtractionPrompt({
|
||||
text,
|
||||
today,
|
||||
@@ -239,7 +242,7 @@ async function handleExtract(request: NextRequest): Promise<Response> {
|
||||
// paying for their own key by the operator's ceiling. `resolveDailyCap`
|
||||
// charges the operator ceiling only when the chain's primary is the
|
||||
// operator's own credential; every other chain gets the user-plan ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = accountToday;
|
||||
const estimatedTokens =
|
||||
MAX_OUTPUT_TOKENS +
|
||||
Math.ceil((systemPrompt.length + userPrompt.length) / 4);
|
||||
|
||||
@@ -41,7 +41,12 @@ const {
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: bossMock }));
|
||||
|
||||
vi.mock("@/lib/oura/client", () => ({ exchangeCode: exchangeMock }));
|
||||
vi.mock("@/lib/oura/client", async (importOriginal) => ({
|
||||
isAcceptedOuraIssuer: (
|
||||
await importOriginal<typeof import("@/lib/oura/client")>()
|
||||
).isAcceptedOuraIssuer,
|
||||
exchangeCode: exchangeMock,
|
||||
}));
|
||||
vi.mock("@/lib/oura/credentials", () => ({
|
||||
getOuraClientCredentials: getCredsMock,
|
||||
}));
|
||||
@@ -65,10 +70,12 @@ function makeReq(opts: {
|
||||
code?: string;
|
||||
state?: string;
|
||||
cookie?: string;
|
||||
iss?: string;
|
||||
}): NextRequest {
|
||||
const params = new URLSearchParams();
|
||||
if (opts.code) params.set("code", opts.code);
|
||||
if (opts.state) params.set("state", opts.state);
|
||||
if (opts.iss) params.set("iss", opts.iss);
|
||||
return {
|
||||
url: `https://app.example/api/oura/callback?${params}`,
|
||||
cookies: {
|
||||
@@ -127,10 +134,11 @@ describe("GET /api/oura/callback", () => {
|
||||
const res = await GET(
|
||||
makeReq({ code: "auth-code", state: "S", cookie: "S" }),
|
||||
);
|
||||
expect(exchangeMock).toHaveBeenCalledWith("auth-code", {
|
||||
clientId: "c",
|
||||
clientSecret: "s",
|
||||
});
|
||||
expect(exchangeMock).toHaveBeenCalledWith(
|
||||
"auth-code",
|
||||
{ clientId: "c", clientSecret: "s" },
|
||||
null,
|
||||
);
|
||||
const data = userUpdate.mock.calls[0]![0].data;
|
||||
expect(data.ouraAccessTokenEncrypted).toBe("enc:at");
|
||||
expect(data.ouraRefreshTokenEncrypted).toBe("enc:rt");
|
||||
@@ -139,6 +147,34 @@ describe("GET /api/oura/callback", () => {
|
||||
});
|
||||
});
|
||||
|
||||
// Oura's current authorization server names itself on the callback
|
||||
// (RFC 9207); the issuer travels on to pick the token endpoint.
|
||||
describe("GET /api/oura/callback — issuer", () => {
|
||||
it("passes Oura's issuer on to the code exchange", async () => {
|
||||
const iss = "https://moi.ouraring.com/oauth/v2/ext/oauth-anonymous";
|
||||
const res = await GET(makeReq({ code: "c", state: "S", cookie: "S", iss }));
|
||||
expect(exchangeMock).toHaveBeenCalledWith(
|
||||
"c",
|
||||
{ clientId: "c", clientSecret: "s" },
|
||||
iss,
|
||||
);
|
||||
expect(res.headers.get("location")).toContain("oura=connected");
|
||||
});
|
||||
|
||||
it("refuses a callback that names a foreign authorization server", async () => {
|
||||
const res = await GET(
|
||||
makeReq({
|
||||
code: "c",
|
||||
state: "S",
|
||||
cookie: "S",
|
||||
iss: "https://evil.example/oauth",
|
||||
}),
|
||||
);
|
||||
expect(res.headers.get("location")).toContain("reason=issuer");
|
||||
expect(exchangeMock).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* v1.32.28 — connecting starts a sync straight away. The poll handler has
|
||||
* accepted a single-user payload since it was written; this callback is the
|
||||
|
||||
@@ -4,7 +4,7 @@ import { getSession } from "@/lib/auth/session";
|
||||
import { annotate, getEvent } from "@/lib/logging/context";
|
||||
import { auditLog } from "@/lib/auth/audit";
|
||||
import { encrypt } from "@/lib/crypto";
|
||||
import { exchangeCode } from "@/lib/oura/client";
|
||||
import { exchangeCode, isAcceptedOuraIssuer } from "@/lib/oura/client";
|
||||
import { getOuraClientCredentials } from "@/lib/oura/credentials";
|
||||
import {
|
||||
oauthStateCookieName,
|
||||
@@ -67,11 +67,20 @@ export const GET = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
if (!code) return ERR("nocode");
|
||||
|
||||
// RFC 9207: an authorization server that names itself must be Oura's. The
|
||||
// issuer also tells which token endpoint the code belongs to. Older apps
|
||||
// send none, which is fine.
|
||||
const iss = searchParams.get("iss");
|
||||
if (!isAcceptedOuraIssuer(iss)) {
|
||||
annotate({ meta: { reason: "issuer" } });
|
||||
return ERR("issuer");
|
||||
}
|
||||
|
||||
try {
|
||||
const creds = await getOuraClientCredentials(userId);
|
||||
if (!creds) return ERR("nocreds");
|
||||
|
||||
const tokens = await exchangeCode(code, creds);
|
||||
const tokens = await exchangeCode(code, creds, iss);
|
||||
|
||||
await prisma.user.update({
|
||||
where: { id: userId },
|
||||
|
||||
@@ -717,6 +717,18 @@ function DaySheet({
|
||||
event.preventDefault();
|
||||
returnFocus();
|
||||
}}
|
||||
onEscapeKeyDown={(event) => {
|
||||
// A dialog opened from inside the day (the date's calendar) is
|
||||
// portaled out of the panel. For a moment after it opens, until
|
||||
// Radix has re-rendered both layers, the panel still holds the
|
||||
// Escape listener: an Escape aimed at the calendar then closed the
|
||||
// whole day. Escape from inside another dialog is never the day's.
|
||||
const owner =
|
||||
event.target instanceof Element
|
||||
? event.target.closest('[role="dialog"]')
|
||||
: null;
|
||||
if (owner && owner !== panelRef.current) event.preventDefault();
|
||||
}}
|
||||
className={cn(
|
||||
// Quicker than the primitive's 500 ms, which read as lag on a
|
||||
// layer opened from a tap; no slide at all under reduced motion.
|
||||
|
||||
@@ -81,3 +81,41 @@ describe("the detail sheet's original-document transport", () => {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* "Open in new tab" hands the browser the types it renders itself, so a long
|
||||
* report gets the browser's zoom and search. It must open the object URL the
|
||||
* sheet already holds (a bare navigation to the route carries no record-session
|
||||
* assertion and is refused once the session has been inside a shared record),
|
||||
* and it must never appear for a type outside the inline serving class.
|
||||
*/
|
||||
describe("the detail sheet's open-in-new-tab control", () => {
|
||||
const control = SHEET.slice(
|
||||
SHEET.indexOf("function DocumentOpenInTabButton("),
|
||||
SHEET.indexOf("/** Inline preview for Class A documents"),
|
||||
);
|
||||
|
||||
it("exists", () => {
|
||||
expect(control.length).toBeGreaterThan(0);
|
||||
expect(SHEET).toMatch(/<DocumentOpenInTabButton\s/);
|
||||
});
|
||||
|
||||
it("renders only for the inline serving class", () => {
|
||||
expect(SHEET).toMatch(
|
||||
/doc\.servingClass === "inline" \? \(\s*\n\s*<DocumentOpenInTabButton\s+readyUrl=\{inlineSrc\}/,
|
||||
);
|
||||
expect([...SHEET.matchAll(/<DocumentOpenInTabButton\s/g)]).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("opens the fetched object URL, never the route, without an opener", () => {
|
||||
expect(control).toContain('window.open(readyUrl, "_blank", "noopener")');
|
||||
expect(control).not.toContain("/api/documents");
|
||||
});
|
||||
|
||||
it("is disabled until the bytes are in hand, and when they failed", () => {
|
||||
expect(control).toMatch(/disabled=\{readyUrl === null \|\| unavailable\}/);
|
||||
expect(SHEET).toMatch(
|
||||
/<DocumentOpenInTabButton\s+readyUrl=\{inlineSrc\}\s+unavailable=\{inlineFailed\}/,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
ArrowRight,
|
||||
Check,
|
||||
Download,
|
||||
ExternalLink,
|
||||
FlaskConical,
|
||||
Pencil,
|
||||
Plus,
|
||||
@@ -199,6 +200,43 @@ function DocumentDownloadButton({
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the stored original in a new browser tab, for the types the browser
|
||||
* renders itself (the inline serving class: PDF, JPEG, PNG, WebP, GIF), so a
|
||||
* long report gets the browser's own zoom and search.
|
||||
*
|
||||
* It opens the object URL the sheet already holds rather than the route: a
|
||||
* plain navigation carries no record-session assertion, so the fence would
|
||||
* refuse it once the session has been inside a shared record. The bytes are
|
||||
* already in hand, so the tab opens inside the click and no popup blocker
|
||||
* steps in. A type outside the inline class never gets the control.
|
||||
*/
|
||||
function DocumentOpenInTabButton({
|
||||
readyUrl,
|
||||
unavailable,
|
||||
}: {
|
||||
readyUrl: string | null;
|
||||
unavailable: boolean;
|
||||
}) {
|
||||
const { t } = useTranslations();
|
||||
return (
|
||||
<Button
|
||||
variant="outline"
|
||||
data-slot="document-open-in-tab"
|
||||
disabled={readyUrl === null || unavailable}
|
||||
aria-label={t("documents.detail.openInNewTab")}
|
||||
onClick={() => {
|
||||
if (readyUrl) window.open(readyUrl, "_blank", "noopener");
|
||||
}}
|
||||
>
|
||||
<ExternalLink className="size-4" aria-hidden />
|
||||
<span className="hidden sm:inline">
|
||||
{t("documents.detail.openInNewTab")}
|
||||
</span>
|
||||
</Button>
|
||||
);
|
||||
}
|
||||
|
||||
/** Inline preview for Class A documents; skeleton until the blob paints. */
|
||||
function InlinePreview({
|
||||
src,
|
||||
@@ -759,8 +797,8 @@ export function DocumentDetailSheet({
|
||||
doc ? (
|
||||
// Delete sits bottom-LEFT, quieted to a text button: the
|
||||
// bottom-right slot reads as the primary/confirm action, and a
|
||||
// destructive control there is a footgun. Share + Download — the
|
||||
// safe, expected actions — keep the trailing edge. On phones their
|
||||
// destructive control there is a footgun. Share, open in a new tab and
|
||||
// Download, the safe, expected actions, keep the trailing edge. On phones their
|
||||
// labels collapse to icon-only (aria-labelled) so three controls
|
||||
// never overflow the sheet at 360 px; the labels return at `sm+`.
|
||||
<div className="flex w-full items-center justify-between gap-2">
|
||||
@@ -824,6 +862,12 @@ export function DocumentDetailSheet({
|
||||
</span>
|
||||
</Button>
|
||||
)}
|
||||
{doc.servingClass === "inline" ? (
|
||||
<DocumentOpenInTabButton
|
||||
readyUrl={inlineSrc}
|
||||
unavailable={inlineFailed}
|
||||
/>
|
||||
) : null}
|
||||
<DocumentDownloadButton
|
||||
documentId={doc.id}
|
||||
filename={doc.filename}
|
||||
|
||||
@@ -699,7 +699,7 @@ describe("<MessageThread>", () => {
|
||||
/>,
|
||||
);
|
||||
expect(html).toContain(
|
||||
"Daily limit reached. The budget refreshes at midnight UTC.",
|
||||
"Daily limit reached. The budget refreshes at midnight in your time zone.",
|
||||
);
|
||||
// v1.4.33 IW7 — fallback copy rewritten ("AI provider" -> "Insights
|
||||
// provider"). Both old and new strings should be absent here when
|
||||
|
||||
@@ -560,7 +560,7 @@ function ChatBubbleImpl({
|
||||
|
||||
// v1.4.25 W5 — map server-emitted error codes to specific Coach
|
||||
// i18n keys. Distinct user-quota and provider-rate-limit copy so the
|
||||
// user understands daily-cap (resets at UTC midnight) vs. transient
|
||||
// user understands daily-cap (resets at their local midnight) vs. transient
|
||||
// provider load (retry in ~5 min). Codes that have no dedicated
|
||||
// translation fall back to the generic provider-unavailable copy.
|
||||
const errorKey = errorCode ? errorCodeToI18nKey(errorCode) : null;
|
||||
|
||||
@@ -28,6 +28,17 @@ export async function register() {
|
||||
console.error("[instrumentation] Encryption key check crashed:", err);
|
||||
}
|
||||
|
||||
// Every connection is pinned to UTC; say clearly if one is not.
|
||||
try {
|
||||
const [{ runBootTimeZoneCheck }, { prisma }] = await Promise.all([
|
||||
import("@/lib/boot/time-zone-check"),
|
||||
import("@/lib/db"),
|
||||
]);
|
||||
await runBootTimeZoneCheck(prisma);
|
||||
} catch {
|
||||
// A diagnostic must never break boot.
|
||||
}
|
||||
|
||||
// Every Node process watches its own event loop — web-only
|
||||
// containers included; a stall report from the process serving
|
||||
// requests is the whole point.
|
||||
|
||||
@@ -7,6 +7,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
buildSessionOptions,
|
||||
getConnectionBudget,
|
||||
pinTimeZoneOnConnect,
|
||||
SET_SESSION_TIME_ZONE,
|
||||
timeZoneSessionOptions,
|
||||
getPgBossPoolMax,
|
||||
getPoolConnectionTimeoutMs,
|
||||
getPrismaPoolMax,
|
||||
@@ -135,14 +138,14 @@ describe("buildSessionOptions", () => {
|
||||
it("keeps work_mem when the timeout is disabled", () => {
|
||||
process.env.DATABASE_STATEMENT_TIMEOUT_MS = "0";
|
||||
delete process.env.DATABASE_WORK_MEM;
|
||||
expect(buildSessionOptions()).toBe("-c work_mem=16MB");
|
||||
expect(buildSessionOptions()).toBe("-c work_mem=16MB -c TimeZone=UTC");
|
||||
});
|
||||
|
||||
it("sets the default work_mem beside the timeouts", () => {
|
||||
delete process.env.DATABASE_STATEMENT_TIMEOUT_MS;
|
||||
delete process.env.DATABASE_WORK_MEM;
|
||||
expect(buildSessionOptions()).toBe(
|
||||
"-c statement_timeout=60000 -c idle_in_transaction_session_timeout=60000 -c work_mem=16MB",
|
||||
"-c statement_timeout=60000 -c idle_in_transaction_session_timeout=60000 -c work_mem=16MB -c TimeZone=UTC",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -166,6 +169,38 @@ describe("buildSessionOptions", () => {
|
||||
});
|
||||
});
|
||||
|
||||
// Every session runs in UTC whatever the database default says: the stored
|
||||
// `timestamp` values are UTC and `now()` reads through the session zone.
|
||||
describe("session time zone pin", () => {
|
||||
afterEach(() => {
|
||||
delete process.env.DATABASE_SESSION_OPTIONS_DISABLED;
|
||||
});
|
||||
|
||||
it("rides the startup options of every pool", () => {
|
||||
expect(buildSessionOptions()).toContain("-c TimeZone=UTC");
|
||||
expect(timeZoneSessionOptions()).toBe("-c TimeZone=UTC");
|
||||
});
|
||||
|
||||
it("adds no connect hook while the startup options carry the pin", () => {
|
||||
const on = vi.fn();
|
||||
pinTimeZoneOnConnect({ on } as never);
|
||||
expect(on).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("pins by statement on connect when the startup options are off", async () => {
|
||||
process.env.DATABASE_SESSION_OPTIONS_DISABLED = "true";
|
||||
expect(timeZoneSessionOptions()).toBeUndefined();
|
||||
const on = vi.fn();
|
||||
pinTimeZoneOnConnect({ on } as never);
|
||||
expect(on).toHaveBeenCalledWith("connect", expect.any(Function));
|
||||
const query = vi.fn().mockResolvedValue(undefined);
|
||||
const hook = on.mock.calls[0][1] as (client: unknown) => void;
|
||||
hook({ query });
|
||||
expect(query).toHaveBeenCalledWith(SET_SESSION_TIME_ZONE);
|
||||
expect(SET_SESSION_TIME_ZONE).toBe("SET TIME ZONE 'UTC'");
|
||||
});
|
||||
});
|
||||
|
||||
describe("process-wide Prisma client", () => {
|
||||
const globalSlot = globalThis as unknown as { prisma?: unknown };
|
||||
|
||||
|
||||
@@ -22,16 +22,45 @@ vi.mock("@/lib/db", () => ({
|
||||
},
|
||||
}));
|
||||
|
||||
const { zoneOf } = vi.hoisted(() => ({
|
||||
zoneOf: vi.fn(async (userId: string) =>
|
||||
userId === "berlin"
|
||||
? "Europe/Berlin"
|
||||
: userId === "new-york"
|
||||
? "America/New_York"
|
||||
: "UTC",
|
||||
),
|
||||
}));
|
||||
vi.mock("@/lib/tz/resolver", () => ({ resolveUserTimezone: zoneOf }));
|
||||
|
||||
/**
|
||||
* The budget day turns at the account's own midnight, the boundary the daily
|
||||
* briefing and the other per-day generations use. On the UTC day, the first
|
||||
* generations of a new local day east of UTC were counted against the old
|
||||
* day's exhausted ceiling.
|
||||
*/
|
||||
describe("buildDateKey", () => {
|
||||
it("formats UTC YYYY-MM-DD", () => {
|
||||
// 2026-05-10T22:30Z is UTC May 10 — Berlin would be May 11 already
|
||||
const key = buildDateKey(new Date("2026-05-10T22:30:00.000Z"));
|
||||
expect(key).toBe("2026-05-10");
|
||||
it("is the account's local day, not the UTC day", async () => {
|
||||
// 22:30 UTC on May 10 is already 00:30 on May 11 in Berlin (CEST).
|
||||
const at = new Date("2026-05-10T22:30:00.000Z");
|
||||
await expect(buildDateKey("berlin", at)).resolves.toBe("2026-05-11");
|
||||
await expect(buildDateKey("utc", at)).resolves.toBe("2026-05-10");
|
||||
expect(zoneOf).toHaveBeenCalledWith("berlin");
|
||||
});
|
||||
|
||||
it("rolls forward at UTC midnight", () => {
|
||||
const key = buildDateKey(new Date("2026-05-11T00:01:00.000Z"));
|
||||
expect(key).toBe("2026-05-11");
|
||||
it("turns at local midnight on either side of it", async () => {
|
||||
// Berlin: 23:59 local on May 10, then 00:01 local on May 11.
|
||||
await expect(
|
||||
buildDateKey("berlin", new Date("2026-05-10T21:59:00.000Z")),
|
||||
).resolves.toBe("2026-05-10");
|
||||
await expect(
|
||||
buildDateKey("berlin", new Date("2026-05-10T22:01:00.000Z")),
|
||||
).resolves.toBe("2026-05-11");
|
||||
// West of UTC the local day ends after the UTC one: 02:00 UTC on May 11
|
||||
// is still the evening of May 10 in New York.
|
||||
await expect(
|
||||
buildDateKey("new-york", new Date("2026-05-11T02:00:00.000Z")),
|
||||
).resolves.toBe("2026-05-10");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
+27
-16
@@ -9,12 +9,13 @@
|
||||
* us add per-day analytics ("avg replies per active user") later
|
||||
* without further migrations.
|
||||
*
|
||||
* `dateKey` is a UTC `YYYY-MM-DD` string. UTC, not Europe/Berlin: the
|
||||
* job runner that may seed off-hour tasks (planned v1.5) needs a
|
||||
* boundary that does not jump on DST. Display layers can format with
|
||||
* the user's timezone.
|
||||
* `dateKey` is a `YYYY-MM-DD` string: the calendar day in the account's own
|
||||
* time zone (`buildDateKey`), the same boundary as the daily briefing and
|
||||
* every other per-day surface.
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import { userDayKey } from "@/lib/tz/format";
|
||||
import { resolveUserTimezone } from "@/lib/tz/resolver";
|
||||
import type { ProviderChainType } from "@/lib/ai/provider-chain";
|
||||
|
||||
/**
|
||||
@@ -215,8 +216,9 @@ export interface DailySpend {
|
||||
|
||||
export async function readDailySpend(
|
||||
userId: string,
|
||||
dateKey: string = buildDateKey(),
|
||||
dateKey?: string,
|
||||
): Promise<DailySpend> {
|
||||
dateKey ??= await buildDateKey(userId);
|
||||
const row = await prisma.coachUsage.findUnique({
|
||||
where: { userId_dateKey: { userId, dateKey } },
|
||||
select: { totalTokens: true, operatorTokens: true },
|
||||
@@ -228,15 +230,24 @@ export async function readDailySpend(
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the UTC day-key for a given clock. Defaults to "now".
|
||||
* The budget day of one account at `at` (default now): the calendar day in
|
||||
* the account's own time zone.
|
||||
*
|
||||
* UTC choice: the budget guards spend against the operator's LLM
|
||||
* bill — the bill cycles at the provider's UTC midnight, so aligning
|
||||
* the local meter to the same boundary keeps reasoning trivial.
|
||||
* Up to v1.42.1 this was the UTC day. The daily briefing and the other
|
||||
* per-day generations run on the account's local day, so in a zone east of
|
||||
* UTC the first ones of a new local day still counted against the old UTC
|
||||
* day, whose ceiling the evening had used up: a burst of refusals right
|
||||
* after local midnight. The day now turns where the person's day turns.
|
||||
*
|
||||
* A caller resolves the key once per request or job and hands the same key
|
||||
* to the reservation and the reconcile, so both land on one row even when
|
||||
* the call spans midnight.
|
||||
*/
|
||||
export function buildDateKey(at: Date = new Date()): string {
|
||||
// eslint-disable-next-line healthlog/no-utc-day-key -- UTC by design: the spend meter follows the provider's UTC billing day (see the doc comment)
|
||||
return at.toISOString().slice(0, 10);
|
||||
export async function buildDateKey(
|
||||
userId: string,
|
||||
at: Date = new Date(),
|
||||
): Promise<string> {
|
||||
return userDayKey(at, await resolveUserTimezone(userId));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -321,12 +332,12 @@ export async function reserveBudget(
|
||||
{ total_tokens: number; operator_tokens: number }[]
|
||||
>`
|
||||
INSERT INTO coach_usage (id, user_id, date_key, total_tokens, operator_tokens, message_count, created_at, updated_at)
|
||||
VALUES (gen_random_uuid()::text, ${userId}, ${dateKey}, ${reserved}, ${operatorReserved}, ${messageBump}, NOW(), NOW())
|
||||
VALUES (gen_random_uuid()::text, ${userId}, ${dateKey}, ${reserved}, ${operatorReserved}, ${messageBump}, (NOW() AT TIME ZONE 'UTC'), (NOW() AT TIME ZONE 'UTC'))
|
||||
ON CONFLICT (user_id, date_key) DO UPDATE SET
|
||||
total_tokens = coach_usage.total_tokens + ${reserved},
|
||||
operator_tokens = coach_usage.operator_tokens + ${operatorReserved},
|
||||
message_count = coach_usage.message_count + ${messageBump},
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
RETURNING total_tokens, operator_tokens
|
||||
`;
|
||||
const totalAfter = Number(rows[0]?.total_tokens ?? reserved);
|
||||
@@ -453,7 +464,7 @@ export async function reconcileSpend(
|
||||
UPDATE coach_usage
|
||||
SET total_tokens = GREATEST(0, total_tokens + ${delta}),
|
||||
operator_tokens = GREATEST(0, operator_tokens + ${operatorDelta}),
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
WHERE user_id = ${userId} AND date_key = ${dateKey}
|
||||
`;
|
||||
}
|
||||
@@ -472,7 +483,7 @@ async function refundReservation(
|
||||
SET total_tokens = GREATEST(0, total_tokens - ${reserved}),
|
||||
operator_tokens = GREATEST(0, operator_tokens - ${operatorReserved}),
|
||||
message_count = GREATEST(0, message_count - ${messageBump}),
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
WHERE user_id = ${userId} AND date_key = ${dateKey}
|
||||
`;
|
||||
}
|
||||
|
||||
@@ -240,7 +240,7 @@ export async function deriveClarifyingQuestions(
|
||||
// serialises them on the row's unique key instead. The cap follows the
|
||||
// chain's cost owner, so a user on their own plan is not held to the
|
||||
// operator-cost ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
QUESTIONS_MAX_TOKENS,
|
||||
|
||||
@@ -68,7 +68,7 @@ export async function reserveTurnBudget(args: {
|
||||
// egresses through the operator's own key or account; a chain on the
|
||||
// person's own plan or key gets the generous user-plan ceiling.
|
||||
const dailyCap = resolveDailyCap(chain);
|
||||
const reqDateKey = buildDateKey();
|
||||
const reqDateKey = await buildDateKey(userId);
|
||||
const firstRound = toolMode
|
||||
? Math.max(1, Math.floor(args.firstRound ?? AI_BUDGETS.coach.maxTokens))
|
||||
: AI_BUDGETS.coach.maxTokens;
|
||||
|
||||
@@ -203,14 +203,14 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
|
||||
next_retry_at, updated_at)
|
||||
VALUES
|
||||
(gen_random_uuid()::text, ${userId}, ${providerType}, 'ok', NULL,
|
||||
0, NOW(), NULL, NULL, NOW())
|
||||
0, (NOW() AT TIME ZONE 'UTC'), NULL, NULL, (NOW() AT TIME ZONE 'UTC'))
|
||||
ON CONFLICT (user_id, provider_type) DO UPDATE SET
|
||||
last_result = 'ok',
|
||||
last_status = NULL,
|
||||
consecutive_failures = 0,
|
||||
last_ok_at = NOW(),
|
||||
last_ok_at = (NOW() AT TIME ZONE 'UTC'),
|
||||
next_retry_at = NULL,
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
`;
|
||||
} catch {
|
||||
// Fire-and-forget — a failed write never blocks the hot path.
|
||||
@@ -240,13 +240,13 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
|
||||
next_retry_at, updated_at)
|
||||
VALUES
|
||||
(gen_random_uuid()::text, ${userId}, ${providerType}, ${result},
|
||||
${status}, 1, NULL, NOW(), NOW() + ${interval}::interval, NOW())
|
||||
${status}, 1, NULL, (NOW() AT TIME ZONE 'UTC'), (NOW() AT TIME ZONE 'UTC') + ${interval}::interval, (NOW() AT TIME ZONE 'UTC'))
|
||||
ON CONFLICT (user_id, provider_type) DO UPDATE SET
|
||||
last_result = ${result},
|
||||
last_status = ${status},
|
||||
consecutive_failures = provider_health.consecutive_failures + 1,
|
||||
last_failure_at = NOW(),
|
||||
next_retry_at = NOW() + CASE
|
||||
last_failure_at = (NOW() AT TIME ZONE 'UTC'),
|
||||
next_retry_at = (NOW() AT TIME ZONE 'UTC') + CASE
|
||||
WHEN ${result} = 'hard_failed'
|
||||
AND provider_health.consecutive_failures + 1 >= ${pauseFrom}::int
|
||||
THEN LEAST(
|
||||
@@ -258,7 +258,7 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = {
|
||||
)
|
||||
ELSE ${interval}::interval
|
||||
END,
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
`;
|
||||
} catch {
|
||||
// Fire-and-forget.
|
||||
|
||||
@@ -662,13 +662,17 @@ describe("computeSummariesSlice", () => {
|
||||
// left with slope90 in v1.37.19).
|
||||
for (const days of ["7 days", "30 days"]) {
|
||||
expect(sql).toContain(
|
||||
`(date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}') AT TIME ZONE 'UTC'`,
|
||||
`m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}')`,
|
||||
);
|
||||
}
|
||||
const anchored = sql.match(
|
||||
/date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30) days'\) AT TIME ZONE 'UTC'/g,
|
||||
/date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30) days'\)/g,
|
||||
);
|
||||
expect(anchored?.length).toBe(4);
|
||||
// The bound stays a naive UTC timestamp like measured_at: converting
|
||||
// it to timestamptz would route the comparison through the session
|
||||
// time zone.
|
||||
expect(sql).not.toMatch(/INTERVAL '(?:7|30) days'\) AT TIME ZONE/);
|
||||
|
||||
// The regression windows must NOT fall back to the bare wall-clock
|
||||
// `NOW() - INTERVAL 'N days'` bound the warm path never uses for the
|
||||
@@ -676,7 +680,9 @@ describe("computeSummariesSlice", () => {
|
||||
// REGR_SLOPE onward) so the avg7/avg30 windows above — which DO stay
|
||||
// wall-clock by design — don't trip the assertion.
|
||||
const slopeBlock = sql.slice(sql.indexOf("REGR_SLOPE"));
|
||||
expect(slopeBlock).not.toMatch(/WHERE m\."measured_at" >= NOW\(\) -/);
|
||||
expect(slopeBlock).not.toMatch(
|
||||
/WHERE m\."measured_at" >= \(?NOW\(\)( AT TIME ZONE 'UTC'\))? -/,
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -134,10 +134,10 @@ export function canonicalMeasurementsCte(
|
||||
onlyType?: MeasurementType,
|
||||
): string {
|
||||
const sinceInner = sinceInterval
|
||||
? `AND "measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
|
||||
? `AND "measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'`
|
||||
: "";
|
||||
const sinceOuter = sinceInterval
|
||||
? `AND mm."measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
|
||||
? `AND mm."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'`
|
||||
: "";
|
||||
// A single-type reader (a series for one metric) scopes both halves to that
|
||||
// type, so the pick scans one metric's rows instead of every row the user
|
||||
|
||||
@@ -174,10 +174,14 @@ function windowedMean(filter: string): string {
|
||||
filter,
|
||||
});
|
||||
}
|
||||
const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`);
|
||||
const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`);
|
||||
const AVG7_SQL = windowedMean(
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`,
|
||||
);
|
||||
const AVG30_SQL = windowedMean(
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
|
||||
);
|
||||
const AVG30_LAST_MONTH_SQL = windowedMean(
|
||||
`m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days'`,
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" < (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
|
||||
);
|
||||
|
||||
interface NarrowAggregateRow {
|
||||
@@ -566,7 +570,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
|
||||
PERCENTILE_CONT(0.5) WITHIN GROUP (
|
||||
ORDER BY m."value"
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
|
||||
)::double precision AS median
|
||||
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
||||
GROUP BY m."type"
|
||||
@@ -1041,17 +1045,16 @@ async function computeFromLiveAggregate(
|
||||
PERCENTILE_CONT(0.5) WITHIN GROUP (
|
||||
ORDER BY m."value"
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
|
||||
)::double precision AS median,
|
||||
-- A2-M2 — anchor the regression windows on the UTC-midnight day
|
||||
-- boundary so this cold-fallback path returns the SAME boundary-day
|
||||
-- membership the warm computeFromRollups composes via
|
||||
-- startOfUtcDay(now - N days) + composeWindowedRegression. The
|
||||
-- expression date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL
|
||||
-- 'N days' is the SQL twin of startOfUtcDay; the trailing AT TIME
|
||||
-- ZONE 'UTC' re-materialises the naive truncation as a UTC instant
|
||||
-- so the measured_at (timestamptz) comparison lands on the same wall
|
||||
-- the rollup bucketStart sits on. The wall-clock NOW() - INTERVAL
|
||||
-- 'N days' is the SQL twin of startOfUtcDay. It stays a naive UTC
|
||||
-- timestamp, like measured_at itself, so the comparison never
|
||||
-- passes through the session time zone. The wall-clock NOW() - INTERVAL
|
||||
-- bound let the boundary day flip in/out of the slope window across
|
||||
-- a warm-to-cold coverage transition (cache-dependent slope on the
|
||||
-- same request); the avg7/30/median windows above stay wall-clock
|
||||
@@ -1060,25 +1063,25 @@ async function computeFromLiveAggregate(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
|
||||
)::double precision AS slope7,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
|
||||
)::double precision AS r2_7,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
|
||||
)::double precision AS slope30,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
|
||||
)::double precision AS r2_30
|
||||
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
||||
GROUP BY m."type"
|
||||
|
||||
@@ -186,7 +186,7 @@ export async function consumeRegistrationChallenge(input: {
|
||||
WHERE id = ${input.challengeId}
|
||||
AND user_id = ${input.userId}
|
||||
AND type = ${expectedType}
|
||||
AND expires_at > NOW()
|
||||
AND expires_at > (NOW() AT TIME ZONE 'UTC')
|
||||
RETURNING challenge, created_at
|
||||
`;
|
||||
|
||||
@@ -225,7 +225,7 @@ export async function claimAuthChallenge(input: {
|
||||
WHERE id = ${input.challengeId}
|
||||
AND type = ${input.type}
|
||||
AND user_id = ${input.userId}
|
||||
AND expires_at > NOW()
|
||||
AND expires_at > (NOW() AT TIME ZONE 'UTC')
|
||||
RETURNING challenge, user_id
|
||||
`
|
||||
: await prisma.$queryRaw<
|
||||
@@ -234,7 +234,7 @@ export async function claimAuthChallenge(input: {
|
||||
DELETE FROM auth_challenges
|
||||
WHERE id = ${input.challengeId}
|
||||
AND type = ${input.type}
|
||||
AND expires_at > NOW()
|
||||
AND expires_at > (NOW() AT TIME ZONE 'UTC')
|
||||
RETURNING challenge, user_id
|
||||
`;
|
||||
if (rows.length !== 1) return null;
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import {
|
||||
checkSessionTimeZone,
|
||||
isUtcTimeZone,
|
||||
runBootTimeZoneCheck,
|
||||
} from "../time-zone-check";
|
||||
|
||||
function clientAnswering(rows: unknown) {
|
||||
return { $queryRawUnsafe: vi.fn().mockResolvedValue(rows) };
|
||||
}
|
||||
|
||||
describe("boot time zone check", () => {
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
|
||||
it("accepts the spellings Postgres reports for UTC", () => {
|
||||
for (const name of ["UTC", "Etc/UTC", "utc", "UCT", "Zulu"]) {
|
||||
expect(isUtcTimeZone(name)).toBe(true);
|
||||
}
|
||||
for (const name of ["Europe/Berlin", "GMT+1", "America/New_York"]) {
|
||||
expect(isUtcTimeZone(name)).toBe(false);
|
||||
}
|
||||
});
|
||||
|
||||
it("stays quiet on a UTC session", async () => {
|
||||
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
const client = clientAnswering([{ TimeZone: "UTC" }]);
|
||||
expect(await runBootTimeZoneCheck(client)).toEqual({
|
||||
state: "utc",
|
||||
timeZone: "UTC",
|
||||
});
|
||||
expect(client.$queryRawUnsafe).toHaveBeenCalledWith("SHOW TimeZone");
|
||||
expect(warn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("warns, naming the zone and the fix, on any other zone", async () => {
|
||||
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
const outcome = await runBootTimeZoneCheck(
|
||||
clientAnswering([{ TimeZone: "Europe/Berlin" }]),
|
||||
);
|
||||
expect(outcome).toEqual({ state: "mismatch", timeZone: "Europe/Berlin" });
|
||||
expect(warn).toHaveBeenCalledTimes(1);
|
||||
expect(warn.mock.calls[0][0]).toContain('"Europe/Berlin"');
|
||||
expect(warn.mock.calls[0][0]).toContain("SET timezone TO 'UTC'");
|
||||
});
|
||||
|
||||
it("never throws when the database cannot answer", async () => {
|
||||
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
const client = {
|
||||
$queryRawUnsafe: vi.fn().mockRejectedValue(new Error("down")),
|
||||
};
|
||||
expect(await checkSessionTimeZone(client)).toEqual({
|
||||
state: "error",
|
||||
message: "down",
|
||||
});
|
||||
await runBootTimeZoneCheck(client);
|
||||
expect(warn.mock.calls[0][0]).toContain("skipped: down");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,75 @@
|
||||
/**
|
||||
* Boot check: does a pooled connection really run in UTC?
|
||||
*
|
||||
* HealthLog pins `TimeZone=UTC` on every connection it opens (see
|
||||
* `@/lib/db-session`). This reads the setting back on one pooled connection
|
||||
* and says so in the log when it is anything else: a pooler that drops the
|
||||
* startup options, a role or proxy that overrides the setting. The expiry
|
||||
* checks of passkey challenges, sessions and one-time links read the stored
|
||||
* UTC timestamps through that zone, so a mismatch shows up as "expired"
|
||||
* straight away. The check never throws and never stops the boot.
|
||||
*/
|
||||
|
||||
const UTC_NAMES = /^(etc\/)?(utc|uct|universal|zulu)$/i;
|
||||
|
||||
export interface TimeZoneCheckClient {
|
||||
$queryRawUnsafe<T = unknown>(query: string): Promise<T>;
|
||||
}
|
||||
|
||||
export type TimeZoneCheckOutcome =
|
||||
| { state: "utc"; timeZone: string }
|
||||
| { state: "mismatch"; timeZone: string }
|
||||
| { state: "error"; message: string };
|
||||
|
||||
export function isUtcTimeZone(name: string): boolean {
|
||||
return UTC_NAMES.test(name.trim());
|
||||
}
|
||||
|
||||
export async function checkSessionTimeZone(
|
||||
client: TimeZoneCheckClient,
|
||||
): Promise<TimeZoneCheckOutcome> {
|
||||
try {
|
||||
const rows =
|
||||
await client.$queryRawUnsafe<Array<Record<string, unknown>>>(
|
||||
"SHOW TimeZone",
|
||||
);
|
||||
const value = rows[0] ? Object.values(rows[0])[0] : undefined;
|
||||
const timeZone = typeof value === "string" ? value : String(value);
|
||||
return isUtcTimeZone(timeZone)
|
||||
? { state: "utc", timeZone }
|
||||
: { state: "mismatch", timeZone };
|
||||
} catch (err) {
|
||||
return {
|
||||
state: "error",
|
||||
message: err instanceof Error ? err.message : String(err),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export function timeZoneMismatchMessage(timeZone: string): string {
|
||||
return (
|
||||
`[boot] The database session runs in time zone "${timeZone}", not UTC. ` +
|
||||
"HealthLog stores times as UTC and pins every connection to UTC, but " +
|
||||
"something between the app and the database overrode it. Expiry checks " +
|
||||
"(passkey challenges, sessions, one-time links) will be off by the zone " +
|
||||
"offset. Fix: ALTER DATABASE <name> SET timezone TO 'UTC', and if a " +
|
||||
"connection pooler drops the startup options, set " +
|
||||
"DATABASE_SESSION_OPTIONS_DISABLED=true so the app pins the zone by " +
|
||||
"statement instead."
|
||||
);
|
||||
}
|
||||
|
||||
/** Run the check and log the outcome. Resolves the outcome for tests. */
|
||||
export async function runBootTimeZoneCheck(
|
||||
client: TimeZoneCheckClient,
|
||||
): Promise<TimeZoneCheckOutcome> {
|
||||
const outcome = await checkSessionTimeZone(client);
|
||||
if (outcome.state === "mismatch") {
|
||||
console.warn(timeZoneMismatchMessage(outcome.timeZone));
|
||||
} else if (outcome.state === "error") {
|
||||
console.warn(
|
||||
`[boot] Database time zone check skipped: ${outcome.message.slice(0, 300)}`,
|
||||
);
|
||||
}
|
||||
return outcome;
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
/**
|
||||
* Session settings shared by every Postgres connection HealthLog opens: the
|
||||
* Prisma pool, pg-boss's pool and the dedicated lock and maintenance clients.
|
||||
* Kept apart from `db.ts` so a side connection can use them without loading
|
||||
* the Prisma client.
|
||||
*/
|
||||
import type { Pool, PoolClient } from "pg";
|
||||
|
||||
import { trackBackgroundTask } from "@/lib/logging/background-tasks";
|
||||
|
||||
/**
|
||||
* Whether the app may send the libpq `options` startup parameter at all.
|
||||
*
|
||||
* PgBouncer refuses a connection that carries a startup parameter it does not
|
||||
* know (`unsupported startup parameter: options`) unless the operator lists
|
||||
* it in `ignore_startup_parameters`. Up to v1.41 such a host could set the
|
||||
* statement timeout to 0, which dropped the whole string; v1.42 always sends
|
||||
* `work_mem`, so that no longer worked. `DATABASE_SESSION_OPTIONS_DISABLED`
|
||||
* (1, true or yes) sends none of the session settings; the pooler or the
|
||||
* database's own defaults then apply.
|
||||
*/
|
||||
export function sessionOptionsDisabled(): boolean {
|
||||
const raw = process.env.DATABASE_SESSION_OPTIONS_DISABLED?.trim();
|
||||
return raw !== undefined && /^(1|true|yes)$/i.test(raw);
|
||||
}
|
||||
|
||||
/**
|
||||
* The session time zone every connection HealthLog opens runs in.
|
||||
*
|
||||
* Almost every `DateTime` column is `timestamp(3)` WITHOUT time zone and
|
||||
* holds UTC. Postgres converts between that type and `now()` /
|
||||
* `CURRENT_TIMESTAMP` / `timestamptz` through the session time zone, and the
|
||||
* driver sends a `Date` parameter as an offset-free UTC string that a
|
||||
* `timestamptz` column reads in the session time zone. On a database whose
|
||||
* default is anything but UTC (a shared instance set to Europe/Berlin, say)
|
||||
* every expiry check then ran hours off, a passkey challenge expired before
|
||||
* the prompt closed, and `timestamptz` writes landed shifted. Pinning the
|
||||
* session removes the dependence for every query at once, whatever the
|
||||
* database or role default says.
|
||||
*/
|
||||
export const SESSION_TIME_ZONE = "UTC";
|
||||
|
||||
export const TIME_ZONE_OPTION = `-c TimeZone=${SESSION_TIME_ZONE}`;
|
||||
|
||||
/**
|
||||
* The statement that pins the time zone where the startup `options` cannot
|
||||
* be sent (`DATABASE_SESSION_OPTIONS_DISABLED`, a connection pooler).
|
||||
* PgBouncer tracks `TimeZone` per client and replays it onto whichever server
|
||||
* connection serves the next transaction, so a `SET` on connect holds in
|
||||
* transaction pooling too.
|
||||
*/
|
||||
export const SET_SESSION_TIME_ZONE = `SET TIME ZONE '${SESSION_TIME_ZONE}'`;
|
||||
|
||||
/** The startup option alone, for the side connections (pg-boss, locks). */
|
||||
export function timeZoneSessionOptions(): string | undefined {
|
||||
return sessionOptionsDisabled() ? undefined : TIME_ZONE_OPTION;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pin the time zone on every new connection of `pool` by statement, when the
|
||||
* startup options are off. pg queues the statement ahead of the first query
|
||||
* the new client is handed out for, so no query runs before it.
|
||||
*/
|
||||
export function pinTimeZoneOnConnect(pool: Pool): void {
|
||||
if (!sessionOptionsDisabled()) return;
|
||||
pool.on("connect", (client: PoolClient) => {
|
||||
trackBackgroundTask(
|
||||
client.query(SET_SESSION_TIME_ZONE).catch((err: unknown) => {
|
||||
// The boot check reports a session that is not in UTC; this names
|
||||
// the cause when the statement itself failed.
|
||||
console.error("[db] Could not pin the session time zone:", err);
|
||||
}),
|
||||
);
|
||||
});
|
||||
}
|
||||
+37
-21
@@ -1,5 +1,19 @@
|
||||
import { Prisma, PrismaClient } from "@/generated/prisma/client";
|
||||
import { PrismaPg } from "@prisma/adapter-pg";
|
||||
import { Pool, type PoolConfig } from "pg";
|
||||
import {
|
||||
pinTimeZoneOnConnect,
|
||||
sessionOptionsDisabled,
|
||||
TIME_ZONE_OPTION,
|
||||
} from "@/lib/db-session";
|
||||
|
||||
export {
|
||||
pinTimeZoneOnConnect,
|
||||
SESSION_TIME_ZONE,
|
||||
SET_SESSION_TIME_ZONE,
|
||||
sessionOptionsDisabled,
|
||||
timeZoneSessionOptions,
|
||||
} from "@/lib/db-session";
|
||||
|
||||
const globalForPrisma = globalThis as unknown as {
|
||||
prisma: PrismaClient | undefined;
|
||||
@@ -108,22 +122,6 @@ export function getWorkMem(): string {
|
||||
return raw && WORK_MEM_SHAPE.test(raw) ? raw : DEFAULT_WORK_MEM;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the app may send the libpq `options` startup parameter at all.
|
||||
*
|
||||
* PgBouncer refuses a connection that carries a startup parameter it does not
|
||||
* know (`unsupported startup parameter: options`) unless the operator lists
|
||||
* it in `ignore_startup_parameters`. Up to v1.41 such a host could set the
|
||||
* statement timeout to 0, which dropped the whole string; v1.42 always sends
|
||||
* `work_mem`, so that no longer worked. `DATABASE_SESSION_OPTIONS_DISABLED`
|
||||
* (1, true or yes) sends none of the session settings; the pooler or the
|
||||
* database's own defaults then apply.
|
||||
*/
|
||||
export function sessionOptionsDisabled(): boolean {
|
||||
const raw = process.env.DATABASE_SESSION_OPTIONS_DISABLED?.trim();
|
||||
return raw !== undefined && /^(1|true|yes)$/i.test(raw);
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the libpq `options` startup string applying the session settings.
|
||||
* Passed straight through `PrismaPg` to the underlying `pg.Pool`, which
|
||||
@@ -143,17 +141,35 @@ export function buildSessionOptions(): string | undefined {
|
||||
timeoutMs > 0
|
||||
? `-c statement_timeout=${timeoutMs} -c idle_in_transaction_session_timeout=${timeoutMs} `
|
||||
: "";
|
||||
return `${timeouts}-c work_mem=${getWorkMem()}`;
|
||||
return `${timeouts}-c work_mem=${getWorkMem()} ${TIME_ZONE_OPTION}`;
|
||||
}
|
||||
|
||||
function createPrismaClient() {
|
||||
const adapter = new PrismaPg({
|
||||
export function prismaPoolConfig(): PoolConfig {
|
||||
return {
|
||||
connectionString: process.env.DATABASE_URL!,
|
||||
max: getPrismaPoolMax(),
|
||||
connectionTimeoutMillis: getPoolConnectionTimeoutMs(),
|
||||
options: buildSessionOptions(),
|
||||
});
|
||||
return new PrismaClient({ adapter });
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The adapter for the process-wide client. The startup options carry the
|
||||
* UTC pin by default, so the adapter builds its own pool from the config.
|
||||
* With the options off (a pooler) the pool is built here instead, to pin the
|
||||
* zone by statement on connect; the adapter then reuses that one pool for
|
||||
* its whole life, a reconnect included.
|
||||
*/
|
||||
export function createPrismaAdapter(): PrismaPg {
|
||||
const config = prismaPoolConfig();
|
||||
if (!sessionOptionsDisabled()) return new PrismaPg(config);
|
||||
const pool = new Pool(config);
|
||||
pinTimeZoneOnConnect(pool);
|
||||
return new PrismaPg(pool);
|
||||
}
|
||||
|
||||
function createPrismaClient() {
|
||||
return new PrismaClient({ adapter: createPrismaAdapter() });
|
||||
}
|
||||
|
||||
// One client per process, in production too. A production build gives every
|
||||
|
||||
@@ -290,7 +290,7 @@ export async function maybeAutoStageLabFacts(
|
||||
// ceiling under the cost owner the auto-index resolved.
|
||||
// v1.38.19 — auto-staging runs off the upload, with no one waiting
|
||||
// on it: a background surface, so half the day's ceiling.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.ocrExtractText.maxTokens,
|
||||
|
||||
@@ -312,7 +312,7 @@ export async function streamFencedReply(
|
||||
await appendMessage({ conversationId, role: "user", content: message });
|
||||
|
||||
// ── Budget reservation (atomic, before the provider call) ──
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.documentChat.maxTokens,
|
||||
|
||||
@@ -242,7 +242,7 @@ async function tryProviderIndex(
|
||||
return { outcome: null, note: null };
|
||||
}
|
||||
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.documentTranscribe.maxTokens,
|
||||
|
||||
@@ -123,7 +123,7 @@ function selectExpression(
|
||||
const cast = `${ref}::timestamp(3)`;
|
||||
// Prisma fills both on create when the row does not carry them.
|
||||
return field === "createdAt" || field === "updatedAt"
|
||||
? `COALESCE(${cast}, CURRENT_TIMESTAMP)`
|
||||
? `COALESCE(${cast}, CURRENT_TIMESTAMP AT TIME ZONE 'UTC')`
|
||||
: cast;
|
||||
}
|
||||
if (kind === "bytea") return `decode(${ref}, 'hex')`;
|
||||
|
||||
@@ -93,7 +93,7 @@ export interface RunBriefingCompletionArgs {
|
||||
* a pinned seed would reproduce the phrasing the re-roll exists to vary.
|
||||
*/
|
||||
seed?: number;
|
||||
/** UTC day-key override; defaults to today. Tests pin it. */
|
||||
/** Budget day-key override; defaults to the account's today. Tests pin it. */
|
||||
dateKey?: string;
|
||||
/**
|
||||
* v1.41 — `daily_briefing` on the calls that may reason (the generation and
|
||||
@@ -122,7 +122,7 @@ export interface RunBriefingCompletionArgs {
|
||||
export async function runBriefingCompletion(
|
||||
args: RunBriefingCompletionArgs,
|
||||
): Promise<RunRawWithFallbackResult> {
|
||||
const dateKey = args.dateKey ?? buildDateKey();
|
||||
const dateKey = args.dateKey ?? (await buildDateKey(args.userId));
|
||||
|
||||
// v1.41 — one resolution per call: the operator's switch and cap, the payer
|
||||
// from the chain's primary. `undefined` is today's wire.
|
||||
|
||||
@@ -145,10 +145,14 @@ function windowedMean(filter: string): string {
|
||||
filter,
|
||||
});
|
||||
}
|
||||
const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`);
|
||||
const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`);
|
||||
const AVG7_SQL = windowedMean(
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`,
|
||||
);
|
||||
const AVG30_SQL = windowedMean(
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
|
||||
);
|
||||
const AVG30_LAST_MONTH_SQL = windowedMean(
|
||||
`m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days'`,
|
||||
`m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" < (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
|
||||
);
|
||||
|
||||
/**
|
||||
@@ -488,7 +492,7 @@ async function buildFromRollups(
|
||||
m."measured_at" AS measured_at
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = $1
|
||||
AND m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
AND m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
|
||||
AND m."deleted_at" IS NULL
|
||||
ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC
|
||||
`,
|
||||
@@ -718,10 +722,9 @@ async function buildFromLiveAggregate(
|
||||
-- composes from (startOfUtcDay(now - N days) +
|
||||
-- composeWindowedRegression). The expression
|
||||
-- date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days'
|
||||
-- is the SQL twin of startOfUtcDay; the trailing AT TIME ZONE 'UTC'
|
||||
-- re-materialises the naive truncation as a UTC instant so the
|
||||
-- measured_at (timestamptz) comparison is on the same wall the
|
||||
-- rollup bucketStart sits on. Without this anchor an account that
|
||||
-- is the SQL twin of startOfUtcDay. It stays a naive UTC timestamp,
|
||||
-- like measured_at itself, so the comparison never passes through
|
||||
-- the session time zone. Without this anchor an account that
|
||||
-- crosses a coverage boundary (warm to cold) saw the boundary day
|
||||
-- flip in/out of the slope window, giving a cache-dependent answer
|
||||
-- to the same request.
|
||||
@@ -729,25 +732,25 @@ async function buildFromLiveAggregate(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
|
||||
)::double precision AS slope7,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
|
||||
)::double precision AS r2_7,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
|
||||
)::double precision AS slope30,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
|
||||
)::double precision AS r2_30
|
||||
FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
|
||||
JOIN window_stats ws ON ws."type" = m."type"
|
||||
@@ -765,7 +768,7 @@ async function buildFromLiveAggregate(
|
||||
m."measured_at" AS measured_at
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = $1
|
||||
AND m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
AND m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
|
||||
AND m."deleted_at" IS NULL
|
||||
ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC
|
||||
`,
|
||||
|
||||
@@ -250,7 +250,7 @@ export async function runStatusCompletion(
|
||||
// credential (`admin-openai` / `admin-codex`). A self-hoster on their own key
|
||||
// or a local model is measured against the generous user-plan ceiling, so
|
||||
// their own hardware/plan is never rationed by the operator's bill.
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
// Thinking is billed as output, so a reasoning call reserves its thinking
|
||||
// budget too; the reconcile below settles on the reported count.
|
||||
const estimatedTokens =
|
||||
|
||||
@@ -70,6 +70,7 @@ describe("web-process pg-boss producer", () => {
|
||||
connectionString: "postgres://healthlog@test/db",
|
||||
max: 2,
|
||||
connectionTimeoutMillis: 20_000,
|
||||
options: "-c TimeZone=UTC",
|
||||
migrate: false,
|
||||
schedule: false,
|
||||
supervise: false,
|
||||
@@ -111,6 +112,7 @@ describe("worker pg-boss pool", () => {
|
||||
connectionString: "postgres://healthlog@test/db",
|
||||
max: 2,
|
||||
connectionTimeoutMillis: 20_000,
|
||||
options: "-c TimeZone=UTC",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -7,24 +7,37 @@
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { prismaMock, syncUserGoogleHealthMock, isParkedMock } = vi.hoisted(
|
||||
() => ({
|
||||
prismaMock: {
|
||||
googleHealthConnection: {
|
||||
findMany: vi.fn(),
|
||||
findUnique: vi.fn(async () => null),
|
||||
update: vi.fn(),
|
||||
},
|
||||
const {
|
||||
prismaMock,
|
||||
syncUserGoogleHealthMock,
|
||||
isParkedMock,
|
||||
bossSend,
|
||||
resumeState,
|
||||
} = vi.hoisted(() => ({
|
||||
prismaMock: {
|
||||
googleHealthConnection: {
|
||||
findMany: vi.fn(),
|
||||
findUnique: vi.fn(
|
||||
async (): Promise<{
|
||||
backfillProgress: unknown;
|
||||
backfillCompletedAt: Date | null;
|
||||
} | null> => null,
|
||||
),
|
||||
update: vi.fn(),
|
||||
},
|
||||
syncUserGoogleHealthMock: vi.fn(),
|
||||
isParkedMock: vi.fn(async () => false),
|
||||
}),
|
||||
);
|
||||
},
|
||||
syncUserGoogleHealthMock: vi.fn(),
|
||||
isParkedMock: vi.fn(async () => false),
|
||||
bossSend: vi.fn(),
|
||||
// Collections the mocked walk "finishes" in this attempt, added to the
|
||||
// resume set the way `runGoogleHealthCollection` adds them.
|
||||
resumeState: { finishes: [] as string[], doneBefore: [] as string[] },
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: () => ({ send: vi.fn() }),
|
||||
getGlobalBoss: () => ({ send: bossSend }),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
@@ -40,12 +53,16 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
|
||||
vi.mock("@/lib/google-health/sync-core", () => ({
|
||||
GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
|
||||
readGoogleHealthBackfillDone: () => new Set<string>(),
|
||||
readGoogleHealthBackfillDone: () => new Set<string>(resumeState.doneBefore),
|
||||
runWithGoogleHealthBackfillResume: async (
|
||||
_userId: string,
|
||||
_done: Set<string>,
|
||||
done: Set<string>,
|
||||
fn: () => Promise<unknown>,
|
||||
) => ({ result: await fn(), skipped: [] }),
|
||||
) => {
|
||||
const result = await fn();
|
||||
for (const key of resumeState.finishes) done.add(key);
|
||||
return { result, skipped: [] };
|
||||
},
|
||||
}));
|
||||
vi.mock("@/lib/google-health/sync", () => ({
|
||||
isGoogleHealthParked: isParkedMock,
|
||||
@@ -63,6 +80,10 @@ beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
isParkedMock.mockResolvedValue(false);
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
prismaMock.googleHealthConnection.findUnique.mockResolvedValue(null);
|
||||
bossSend.mockResolvedValue("job-next");
|
||||
resumeState.finishes = [];
|
||||
resumeState.doneBefore = [];
|
||||
});
|
||||
|
||||
describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
|
||||
@@ -150,3 +171,92 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
|
||||
expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* A long history walks in slices. A slice that finished more collections than
|
||||
* it started with is progressing, not failing: it queues the next slice and
|
||||
* ends quietly, without the error line or the failure count. Only a slice that
|
||||
* finished nothing new is stuck and keeps throwing.
|
||||
*/
|
||||
describe("runGoogleHealthBackfillForUser — slices", () => {
|
||||
it("continues a progressing slice instead of failing it", async () => {
|
||||
resumeState.doneBefore = ["steps"];
|
||||
resumeState.finishes = ["sleep", "weight"];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 9, failed: true });
|
||||
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser("u1", undefined, 2),
|
||||
).resolves.toEqual({
|
||||
imported: 9,
|
||||
continued: true,
|
||||
});
|
||||
|
||||
expect(bossSend).toHaveBeenCalledTimes(1);
|
||||
const [queue, payload, options] = bossSend.mock.calls[0]!;
|
||||
expect(queue).toBe("integration-backfill-admission");
|
||||
expect(payload).toMatchObject({
|
||||
kind: "google-health-backfill",
|
||||
data: { userId: "u1", slice: 3 },
|
||||
});
|
||||
// Its own key, so the exclusive policy does not swallow it while this
|
||||
// slice is still active.
|
||||
expect(options).toMatchObject({
|
||||
singletonKey: "google-health-backfill|u1|slice-3",
|
||||
startAfter: expect.any(Number),
|
||||
});
|
||||
// Not complete: no stamp.
|
||||
expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("still throws for a slice that finished nothing new", async () => {
|
||||
resumeState.doneBefore = ["steps"];
|
||||
resumeState.finishes = [];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 3, failed: true });
|
||||
|
||||
await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow(
|
||||
/incomplete/,
|
||||
);
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("throws rather than drop the work when the next slice cannot be queued", async () => {
|
||||
resumeState.finishes = ["sleep"];
|
||||
bossSend.mockResolvedValue(null);
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 1, failed: true });
|
||||
|
||||
await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow(
|
||||
/incomplete/,
|
||||
);
|
||||
});
|
||||
|
||||
it("stamps the marker on the slice that finishes the history", async () => {
|
||||
resumeState.doneBefore = ["steps", "sleep"];
|
||||
resumeState.finishes = ["weight"];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 4, failed: false });
|
||||
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser("u1", undefined, 5),
|
||||
).resolves.toEqual({
|
||||
imported: 4,
|
||||
});
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
expect(
|
||||
prismaMock.googleHealthConnection.update.mock.calls[0]![0].data
|
||||
.backfillCompletedAt,
|
||||
).toBeInstanceOf(Date);
|
||||
});
|
||||
|
||||
it("walks nothing again once an earlier slice completed the import", async () => {
|
||||
prismaMock.googleHealthConnection.findUnique.mockResolvedValue({
|
||||
backfillProgress: null,
|
||||
backfillCompletedAt: new Date(),
|
||||
});
|
||||
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser("u1", undefined, 4),
|
||||
).resolves.toEqual({
|
||||
imported: 0,
|
||||
});
|
||||
expect(syncUserGoogleHealthMock).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -26,6 +26,7 @@ describe("maintenance session settings behind a pooler", () => {
|
||||
"SET maintenance_work_mem = '128MB'",
|
||||
"SET max_parallel_maintenance_workers = '0'",
|
||||
"SET lock_timeout = '300000'",
|
||||
"SET TimeZone = 'UTC'",
|
||||
]);
|
||||
});
|
||||
it("keeps maintenance in one process so it needs no /dev/shm segment", () => {
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
import { PgBoss } from "pg-boss";
|
||||
|
||||
import { getPgBossPoolMax, getPoolConnectionTimeoutMs } from "@/lib/db";
|
||||
import { timeZoneSessionOptions } from "@/lib/db-session";
|
||||
|
||||
const BOSS_KEY = "__healthlog_pgboss__" as const;
|
||||
const BOSS_START_KEY = "__healthlog_pgboss_start__" as const;
|
||||
@@ -76,6 +77,9 @@ function databaseOptions(connectionString: string) {
|
||||
connectionString,
|
||||
max: getPgBossPoolMax(),
|
||||
connectionTimeoutMillis: getPoolConnectionTimeoutMs(),
|
||||
// pg-boss keeps its own tables in `timestamptz`, so its queue math does
|
||||
// not depend on the zone; the pin keeps every HealthLog session alike.
|
||||
options: timeZoneSessionOptions(),
|
||||
} as const;
|
||||
}
|
||||
|
||||
|
||||
@@ -211,7 +211,7 @@ export const composeNudgeWithAI: ComposeNudgeWithAI = async (params) => {
|
||||
const maxTokens = budget.maxTokens ?? 160;
|
||||
// v1.38.19 — a background surface: half the day's ceiling.
|
||||
const dailyCap = resolveDailyCapFor("job", chain);
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(params.userId);
|
||||
|
||||
// Per-user budget gate (atomic reserve; refunds itself on refusal).
|
||||
const reservation = await reserveBudget(
|
||||
|
||||
@@ -147,7 +147,7 @@ async function claimReaction(
|
||||
),
|
||||
operator_tokens = GREATEST(0, operator_tokens - ${operatorReserved}),
|
||||
message_count = GREATEST(0, message_count - 1),
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
WHERE user_id = ${arrival.userId}
|
||||
AND date_key = ${current.generationBudgetDateKey}
|
||||
`;
|
||||
|
||||
@@ -268,7 +268,7 @@ export async function runContentIndexBackfillForUser(
|
||||
break outer;
|
||||
}
|
||||
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.documentTranscribe.maxTokens,
|
||||
|
||||
@@ -223,7 +223,7 @@ export async function runDocumentSummaryJob(
|
||||
return;
|
||||
}
|
||||
|
||||
const dateKey = buildDateKey();
|
||||
const dateKey = await buildDateKey(userId);
|
||||
const reservation = await reserveBudget(
|
||||
userId,
|
||||
AI_BUDGETS.documentSummary.maxTokens,
|
||||
|
||||
@@ -15,7 +15,11 @@ import { Prisma } from "@/generated/prisma/client";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
|
||||
import {
|
||||
enqueueIntegrationBackfillAdmission,
|
||||
INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
integrationBackfillSourceOptions,
|
||||
} from "@/lib/jobs/integration-backfill-admission";
|
||||
import {
|
||||
readGoogleHealthBackfillDone,
|
||||
runWithGoogleHealthBackfillResume,
|
||||
@@ -68,6 +72,14 @@ export interface GoogleHealthBackfillPayload {
|
||||
* (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp
|
||||
* lands, and on every reconnect.
|
||||
*
|
||||
* Runs in slices: an attempt that ends before the whole history is in but
|
||||
* finished at least one more collection than it started with made progress.
|
||||
* It is not a failure: it sends the next slice (`slice + 1`) and returns
|
||||
* `continued`, without the error line or the queue's failure count. Only an
|
||||
* attempt that finished nothing new throws, because that one is stuck: the
|
||||
* pg-boss retry and, after it, the next boot take it from there. Progress is
|
||||
* bounded by the number of collections, so the chain of slices is too.
|
||||
*
|
||||
* Never beside another run of the same account: the sync takes the account's
|
||||
* lock, waiting a bounded time for an hourly or manual run to finish. If the
|
||||
* lock does not come free the attempt throws without doing anything, and the
|
||||
@@ -76,7 +88,8 @@ export interface GoogleHealthBackfillPayload {
|
||||
export async function runGoogleHealthBackfillForUser(
|
||||
userId: string,
|
||||
shouldStop: () => boolean = () => false,
|
||||
): Promise<{ imported: number }> {
|
||||
slice: number = 0,
|
||||
): Promise<{ imported: number; continued?: boolean }> {
|
||||
if (await isGoogleHealthParked(userId)) {
|
||||
annotate({
|
||||
action: {
|
||||
@@ -89,9 +102,21 @@ export async function runGoogleHealthBackfillForUser(
|
||||
|
||||
const connection = await prisma.googleHealthConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { backfillProgress: true },
|
||||
select: { backfillProgress: true, backfillCompletedAt: true },
|
||||
});
|
||||
if (connection?.backfillCompletedAt) {
|
||||
// A slice queued behind the one that finished the import, or a boot
|
||||
// delivery that raced it: the history is in, nothing to walk again.
|
||||
annotate({
|
||||
action: {
|
||||
name: "google_health.backfill.already_complete",
|
||||
details: { slice },
|
||||
},
|
||||
});
|
||||
return { imported: 0 };
|
||||
}
|
||||
const done = readGoogleHealthBackfillDone(connection?.backfillProgress);
|
||||
const doneBefore = done.size;
|
||||
|
||||
const {
|
||||
result: { imported, failed, busy },
|
||||
@@ -115,9 +140,46 @@ export async function runGoogleHealthBackfillForUser(
|
||||
}
|
||||
|
||||
if (failed) {
|
||||
// Surface through the pg-boss retry path (retryLimit 3, backoff). If the
|
||||
// failure parked the connection at error_reauth meanwhile, the next
|
||||
// attempt takes the reauth return above instead of failing again.
|
||||
// `done` is the resume set this attempt added to as it finished
|
||||
// collections; a larger set than it started with is progress.
|
||||
const finished = done.size - doneBefore;
|
||||
if (finished > 0) {
|
||||
const boss = getGlobalBoss();
|
||||
const sent = boss
|
||||
? await enqueueIntegrationBackfillAdmission(
|
||||
boss,
|
||||
{
|
||||
kind: "google-health-backfill",
|
||||
data: {
|
||||
userId,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
slice: slice + 1,
|
||||
},
|
||||
},
|
||||
INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
)
|
||||
: null;
|
||||
if (sent) {
|
||||
annotate({
|
||||
action: {
|
||||
name: "google_health.backfill.continued",
|
||||
details: {
|
||||
imported,
|
||||
slice,
|
||||
collectionsFinished: finished,
|
||||
collectionsDone: done.size,
|
||||
},
|
||||
},
|
||||
});
|
||||
return { imported, continued: true };
|
||||
}
|
||||
// No follow-up could be queued; fall through to the retry path so the
|
||||
// work is not dropped.
|
||||
}
|
||||
// No collection finished in this attempt: the backfill is stuck, not
|
||||
// slow. Surface through the pg-boss retry path (retryLimit 3, backoff).
|
||||
// If the failure parked the connection at error_reauth meanwhile, the
|
||||
// next attempt takes the reauth return above instead of failing again.
|
||||
throw new Error(
|
||||
`google-health backfill incomplete for user ${userId} — marker not stamped`,
|
||||
);
|
||||
|
||||
@@ -28,6 +28,14 @@ export type IntegrationBackfillData = {
|
||||
userId: string;
|
||||
enqueuedAt: string;
|
||||
provider?: "WHOOP" | "WITHINGS";
|
||||
/**
|
||||
* The continuation number of an import that runs in slices: absent (0) for
|
||||
* the job discovery or a connect sends, 1, 2, ... for each follow-up a
|
||||
* slice that made progress sent for the rest. Part of the job's singleton
|
||||
* key, so the follow-up is not swallowed by the `exclusive` policy while the
|
||||
* slice that sends it is still active; never part of the import identity.
|
||||
*/
|
||||
slice?: number;
|
||||
};
|
||||
|
||||
export interface IntegrationBackfillAdmissionPayload {
|
||||
@@ -88,7 +96,12 @@ export function integrationBackfillSourceOptions(
|
||||
};
|
||||
}
|
||||
|
||||
export function integrationBackfillAdmissionSingletonKey(
|
||||
/**
|
||||
* The import a job belongs to: one kind for one account (and provider). The
|
||||
* import lock keys on it, so no two slices or deliveries of one import ever
|
||||
* run at once.
|
||||
*/
|
||||
export function integrationBackfillIdentity(
|
||||
payload: IntegrationBackfillAdmissionPayload,
|
||||
): string {
|
||||
if (payload.kind === "sleep-timeline-backfill") {
|
||||
@@ -100,17 +113,29 @@ export function integrationBackfillAdmissionSingletonKey(
|
||||
return `${payload.kind}|${payload.data.userId}`;
|
||||
}
|
||||
|
||||
/** The job's singleton key: the import identity, plus its slice number. */
|
||||
export function integrationBackfillAdmissionSingletonKey(
|
||||
payload: IntegrationBackfillAdmissionPayload,
|
||||
): string {
|
||||
const identity = integrationBackfillIdentity(payload);
|
||||
const slice = payload.data.slice ?? 0;
|
||||
return slice > 0 ? `${identity}|slice-${slice}` : identity;
|
||||
}
|
||||
|
||||
export async function enqueueIntegrationBackfillAdmission(
|
||||
boss: BossSender,
|
||||
payload: IntegrationBackfillAdmissionPayload,
|
||||
startAfterSeconds: number = 0,
|
||||
): Promise<string | null> {
|
||||
return boss.send(
|
||||
INTEGRATION_BACKFILL_ADMISSION_QUEUE,
|
||||
payload,
|
||||
integrationBackfillAdmissionSendOptions(payload),
|
||||
);
|
||||
return boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, {
|
||||
...integrationBackfillAdmissionSendOptions(payload),
|
||||
...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
/** Breathing room between one slice of an import and the next. */
|
||||
export const INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS = 60;
|
||||
|
||||
/**
|
||||
* The send options of one admission job. The options ride on the send, not on
|
||||
* `createQueue`, whose `ON CONFLICT DO NOTHING` leaves an existing queue's
|
||||
|
||||
@@ -25,7 +25,7 @@ import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill";
|
||||
import { runSleepTimelineBackfillForUser } from "@/lib/jobs/sleep-timeline-backfill";
|
||||
import { runLabBiomarkerBackfillForUser } from "@/lib/jobs/lab-biomarker-backfill";
|
||||
import {
|
||||
integrationBackfillAdmissionSingletonKey,
|
||||
integrationBackfillIdentity,
|
||||
type IntegrationBackfillAdmissionPayload,
|
||||
} from "@/lib/jobs/integration-backfill-admission";
|
||||
import { withIntegrationBackfillLock } from "@/lib/jobs/integration-backfill-lock";
|
||||
@@ -36,6 +36,8 @@ import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
|
||||
interface RunCounts {
|
||||
imported: number;
|
||||
/** Imports that ended a slice with progress and sent the next one. */
|
||||
continued: number;
|
||||
removed: number;
|
||||
deleted: number;
|
||||
markers: number;
|
||||
@@ -66,14 +68,16 @@ async function runAdmitted(
|
||||
return;
|
||||
}
|
||||
case "google-health-backfill": {
|
||||
const { imported } = await runGoogleHealthBackfillForUser(
|
||||
const { imported, continued } = await runGoogleHealthBackfillForUser(
|
||||
userId,
|
||||
shouldStop,
|
||||
data.slice ?? 0,
|
||||
);
|
||||
totals.imported += imported;
|
||||
if (continued) totals.continued += 1;
|
||||
workerLog(
|
||||
"info",
|
||||
`[google-health-backfill] user=${userId} imported=${imported}`,
|
||||
`[google-health-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
@@ -152,6 +156,7 @@ export async function drainIntegrationBackfillAdmission(
|
||||
const shouldStop = jobBudget(jobs);
|
||||
const totals: RunCounts = {
|
||||
imported: 0,
|
||||
continued: 0,
|
||||
removed: 0,
|
||||
deleted: 0,
|
||||
markers: 0,
|
||||
@@ -159,7 +164,7 @@ export async function drainIntegrationBackfillAdmission(
|
||||
};
|
||||
let alreadyRunning = 0;
|
||||
for (const job of jobs) {
|
||||
const identity = integrationBackfillAdmissionSingletonKey(job.data);
|
||||
const identity = integrationBackfillIdentity(job.data);
|
||||
const run = await withIntegrationBackfillLock(identity, () =>
|
||||
runAdmitted(job.data, shouldStop, totals),
|
||||
);
|
||||
@@ -183,6 +188,7 @@ export async function drainIntegrationBackfillAdmission(
|
||||
jobs: jobs.length,
|
||||
already_running: alreadyRunning,
|
||||
imported: totals.imported,
|
||||
continued: totals.continued,
|
||||
removed: totals.removed,
|
||||
deleted: totals.deleted,
|
||||
markers: totals.markers,
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
* One extra connection per running locked job.
|
||||
*/
|
||||
import { Client } from "pg";
|
||||
import { timeZoneSessionOptions } from "@/lib/db-session";
|
||||
import { caughtAs } from "@/lib/logging/signal";
|
||||
|
||||
/** What a guarded run came to. */
|
||||
@@ -50,7 +51,11 @@ export async function withJobLock<T>(
|
||||
connectionString: string | undefined = process.env.DATABASE_URL,
|
||||
options: JobLockOptions = {},
|
||||
): Promise<GuardedRun<T>> {
|
||||
const client = new Client({ connectionString, keepAlive: true });
|
||||
const client = new Client({
|
||||
connectionString,
|
||||
keepAlive: true,
|
||||
options: timeZoneSessionOptions(),
|
||||
});
|
||||
// An idle client whose server goes away emits `error`; unhandled, that
|
||||
// takes the whole worker down. The run itself notices the database is
|
||||
// gone through its own queries.
|
||||
|
||||
@@ -77,7 +77,7 @@ export interface MeasurementMaintenancePayload {
|
||||
* whatever its container settings.
|
||||
*/
|
||||
export const MAINTENANCE_SESSION_OPTIONS =
|
||||
"-c statement_timeout=0 -c idle_in_transaction_session_timeout=0 -c maintenance_work_mem=128MB -c max_parallel_maintenance_workers=0 -c lock_timeout=300000";
|
||||
"-c statement_timeout=0 -c idle_in_transaction_session_timeout=0 -c maintenance_work_mem=128MB -c max_parallel_maintenance_workers=0 -c lock_timeout=300000 -c TimeZone=UTC";
|
||||
|
||||
/**
|
||||
* The same settings as `SET` statements, for a host behind a connection
|
||||
|
||||
@@ -592,7 +592,7 @@ export async function runReactionLine(
|
||||
generationClaimId,
|
||||
job.userId,
|
||||
revision,
|
||||
buildDateKey(),
|
||||
await buildDateKey(job.userId),
|
||||
// v1.38.19 — a background surface: half the day's ceiling.
|
||||
resolveDailyCapFor("job", chain),
|
||||
resolveCostOwner(chain),
|
||||
|
||||
@@ -294,7 +294,7 @@ export async function handleRateLimitCleanup(
|
||||
const p = getWorkerPrisma();
|
||||
try {
|
||||
const result = await p.$executeRaw`
|
||||
DELETE FROM rate_limits WHERE reset_at < NOW()
|
||||
DELETE FROM rate_limits WHERE reset_at < (NOW() AT TIME ZONE 'UTC')
|
||||
`;
|
||||
evt.addMeta("rate_limit_cleanup_deleted", result);
|
||||
return jobDone({ deleted: result });
|
||||
|
||||
@@ -274,7 +274,7 @@ export async function enqueueBootTimeStepConsolidationRepair(
|
||||
FROM measurements t
|
||||
WHERE t."type" = 'ACTIVITY_STEPS'
|
||||
AND t."deleted_at" IS NOT NULL
|
||||
AND t."deleted_at" >= now() - make_interval(days => ${TOMBSTONE_RETENTION_DAYS})
|
||||
AND t."deleted_at" >= (NOW() AT TIME ZONE 'UTC') - make_interval(days => ${TOMBSTONE_RETENTION_DAYS})
|
||||
AND t."source" IN ('GOOGLE_HEALTH', 'FITBIT')
|
||||
AND t."external_id" LIKE 'stats:%'
|
||||
AND NOT EXISTS (
|
||||
|
||||
@@ -326,7 +326,7 @@ export async function applyAppleHealthDistanceRepair(
|
||||
SET value = value * ${factor},
|
||||
unit = 'm',
|
||||
sync_version = sync_version + 1,
|
||||
updated_at = NOW()
|
||||
updated_at = (NOW() AT TIME ZONE 'UTC')
|
||||
WHERE id = ANY(${chunk})`;
|
||||
}
|
||||
await auditLog(APPLE_HEALTH_DISTANCE_REPAIR_ACTION, {
|
||||
|
||||
@@ -166,7 +166,7 @@ export async function claimManagedGuardianEgressAuthorization(
|
||||
${delivery.recipientUserId},
|
||||
${channel},
|
||||
${eventType},
|
||||
clock_timestamp()
|
||||
(clock_timestamp() AT TIME ZONE 'UTC')
|
||||
WHERE EXISTS (
|
||||
SELECT 1
|
||||
FROM "account_grants"
|
||||
@@ -177,7 +177,7 @@ export async function claimManagedGuardianEgressAuthorization(
|
||||
AND "revoked_at" IS NULL
|
||||
AND (
|
||||
"expires_at" IS NULL
|
||||
OR "expires_at" > clock_timestamp()
|
||||
OR "expires_at" > (clock_timestamp() AT TIME ZONE 'UTC')
|
||||
)
|
||||
)
|
||||
RETURNING "id"
|
||||
|
||||
@@ -1036,7 +1036,7 @@ export const medicationExtractRequest = z
|
||||
.regex(/^\d{4}-\d{2}-\d{2}$/)
|
||||
.optional()
|
||||
.describe(
|
||||
'Optional override of the reference date used to resolve relative phrases ("tomorrow", "next Monday"). Format: `YYYY-MM-DD`. Defaults to the server\'s UTC day.',
|
||||
'Optional override of the reference date used to resolve relative phrases ("tomorrow", "next Monday"). Format: `YYYY-MM-DD`. Defaults to the account\'s own day in its time zone.',
|
||||
),
|
||||
})
|
||||
.meta({
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import {
|
||||
OURA_OAUTH_LEGACY_TOKEN_URL,
|
||||
OURA_OAUTH_SCOPE,
|
||||
OURA_OAUTH_TOKEN_URL,
|
||||
RESILIENCE_LEVELS,
|
||||
derivePeriodDaysFromCyclePhases,
|
||||
exchangeCode,
|
||||
@@ -12,6 +14,7 @@ import {
|
||||
fetchVo2Max,
|
||||
getAuthorizationUrl,
|
||||
getOuraCredentials,
|
||||
isAcceptedOuraIssuer,
|
||||
mapCardiovascularAge,
|
||||
mapDailyActivity,
|
||||
mapDailySleep,
|
||||
@@ -132,6 +135,133 @@ describe("exchangeCode / refreshAccessToken", () => {
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Oura moved its authorization server: apps from the current developer
|
||||
* portal get codes from moi.ouraring.com, which the documented
|
||||
* api.ouraring.com token endpoint refuses with 400 invalid_request. A code
|
||||
* works once, so the live endpoint is always asked first.
|
||||
*/
|
||||
describe("token endpoint order", () => {
|
||||
const MOI_ISS = "https://moi.ouraring.com/oauth/v2/ext/oauth-anonymous";
|
||||
const OK = {
|
||||
status: 200,
|
||||
body: { access_token: "a", refresh_token: "r", expires_in: 86400 },
|
||||
};
|
||||
const urls = (mock: ReturnType<typeof installFetchMock>) =>
|
||||
mock.mock.calls.map(([url]) => url);
|
||||
|
||||
it("exchanges a code at the live endpoint, sending the secret in the body", async () => {
|
||||
const fetchMock = installFetchMock([OK]);
|
||||
await exchangeCode("code", CREDS, MOI_ISS);
|
||||
expect(OURA_OAUTH_TOKEN_URL).toBe(
|
||||
"https://moi.ouraring.com/oauth/v2/ext/oauth-token",
|
||||
);
|
||||
expect(urls(fetchMock)).toEqual([OURA_OAUTH_TOKEN_URL]);
|
||||
const body = String(fetchMock.mock.calls[0]![1]!.body);
|
||||
expect(body).toContain("grant_type=authorization_code");
|
||||
expect(body).toContain("client_id=cid");
|
||||
expect(body).toContain("client_secret=csecret");
|
||||
});
|
||||
|
||||
it("never hands a code from the current server to the legacy endpoint", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 400, body: { error: "invalid_request" } },
|
||||
OK,
|
||||
]);
|
||||
await expect(exchangeCode("code", CREDS, MOI_ISS)).rejects.toMatchObject({
|
||||
upstreamError: "invalid_request",
|
||||
});
|
||||
expect(urls(fetchMock)).toEqual([OURA_OAUTH_TOKEN_URL]);
|
||||
expect(urls(fetchMock)).not.toContain(OURA_OAUTH_LEGACY_TOKEN_URL);
|
||||
});
|
||||
|
||||
it("asks the legacy endpoint only after the live one refused an older app's code", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 401, body: { error: "invalid_client" } },
|
||||
OK,
|
||||
]);
|
||||
const t = await exchangeCode("code", CREDS);
|
||||
expect(t.access_token).toBe("a");
|
||||
expect(urls(fetchMock)).toEqual([
|
||||
OURA_OAUTH_TOKEN_URL,
|
||||
OURA_OAUTH_LEGACY_TOKEN_URL,
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not try the legacy endpoint once the live one accepted the code", async () => {
|
||||
const fetchMock = installFetchMock([OK]);
|
||||
await exchangeCode("code", CREDS);
|
||||
expect(urls(fetchMock)).toEqual([OURA_OAUTH_TOKEN_URL]);
|
||||
});
|
||||
|
||||
it("refreshes at the live endpoint first", async () => {
|
||||
const fetchMock = installFetchMock([OK]);
|
||||
await refreshAccessToken("r1", CREDS);
|
||||
expect(urls(fetchMock)).toEqual([OURA_OAUTH_TOKEN_URL]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[400, { error: "invalid_request" }],
|
||||
[404, null],
|
||||
])(
|
||||
"falls back to the legacy endpoint on a %i from the live one",
|
||||
async (status, body) => {
|
||||
const fetchMock = installFetchMock([{ status, body }, OK]);
|
||||
const t = await refreshAccessToken("r1", CREDS);
|
||||
expect(t.refresh_token).toBe("r");
|
||||
expect(urls(fetchMock)).toEqual([
|
||||
OURA_OAUTH_TOKEN_URL,
|
||||
OURA_OAUTH_LEGACY_TOKEN_URL,
|
||||
]);
|
||||
},
|
||||
);
|
||||
|
||||
it("reports the live endpoint's refusal when both refuse", async () => {
|
||||
installFetchMock([
|
||||
{ status: 400, body: { error: "invalid_grant" } },
|
||||
{ status: 400, body: { error: "invalid_request" } },
|
||||
]);
|
||||
await expect(refreshAccessToken("dead", CREDS)).rejects.toMatchObject({
|
||||
upstreamError: "invalid_grant",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not fall back on a server error or a rate limit", async () => {
|
||||
for (const status of [429, 503]) {
|
||||
const fetchMock = installFetchMock([{ status }, OK]);
|
||||
await expect(refreshAccessToken("r1", CREDS)).rejects.toBeInstanceOf(
|
||||
OuraApiError,
|
||||
);
|
||||
expect(urls(fetchMock)).toEqual([OURA_OAUTH_TOKEN_URL]);
|
||||
vi.unstubAllGlobals();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("isAcceptedOuraIssuer", () => {
|
||||
it("accepts no issuer and Oura's own servers", () => {
|
||||
expect(isAcceptedOuraIssuer(null)).toBe(true);
|
||||
expect(
|
||||
isAcceptedOuraIssuer(
|
||||
"https://moi.ouraring.com/oauth/v2/ext/oauth-anonymous",
|
||||
),
|
||||
).toBe(true);
|
||||
expect(isAcceptedOuraIssuer("https://cloud.ouraring.com")).toBe(true);
|
||||
});
|
||||
|
||||
it("refuses anything else", () => {
|
||||
for (const iss of [
|
||||
"https://evil.example/oauth",
|
||||
"http://moi.ouraring.com/oauth",
|
||||
"https://moi.ouraring.com.evil.example/",
|
||||
"not a url",
|
||||
"",
|
||||
]) {
|
||||
expect(isAcceptedOuraIssuer(iss), iss).toBe(false);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("fetchReadiness pagination", () => {
|
||||
it("walks next_token across pages", async () => {
|
||||
installFetchMock([
|
||||
|
||||
+139
-10
@@ -10,7 +10,9 @@
|
||||
*
|
||||
* Oura specifics: standard authorization-code OAuth WITH refresh tokens (the
|
||||
* sync layer refreshes on 401, persisting BOTH rotated tokens). The token
|
||||
* endpoint takes client_id/secret in the body. Collections are read by
|
||||
* endpoint takes client_id/secret in the body (client_secret_post) and lives
|
||||
* on moi.ouraring.com, with the documented api.ouraring.com endpoint as a
|
||||
* fallback for older apps. Collections are read by
|
||||
* `start_date` / `end_date` (YYYY-MM-DD) and page via `next_token`.
|
||||
*/
|
||||
import { getEvent } from "@/lib/logging/context";
|
||||
@@ -22,7 +24,63 @@ import { envOr, envValue } from "@/lib/env";
|
||||
|
||||
const OURA_API_BASE = "https://api.ouraring.com";
|
||||
const OURA_OAUTH_AUTH_URL = "https://cloud.ouraring.com/oauth/authorize";
|
||||
const OURA_OAUTH_TOKEN_URL = "https://api.ouraring.com/oauth/token";
|
||||
|
||||
/**
|
||||
* Oura's live token endpoint. Apps created in the current developer portal
|
||||
* get their codes from this authorization server (the callback carries
|
||||
* `iss=https://moi.ouraring.com/oauth/v2/ext/oauth-anonymous`), and the
|
||||
* documented `api.ouraring.com/oauth/token` answers those codes with
|
||||
* `400 invalid_request`. Older apps are served here too.
|
||||
*/
|
||||
export const OURA_OAUTH_TOKEN_URL =
|
||||
"https://moi.ouraring.com/oauth/v2/ext/oauth-token";
|
||||
|
||||
/**
|
||||
* The documented endpoint, kept only as a fallback for the older apps the
|
||||
* live endpoint turns away. Never tried first for a code: an authorization
|
||||
* code works once (RFC 6749 4.1.2), so a rejected first attempt would burn
|
||||
* it before the right endpoint saw it.
|
||||
*/
|
||||
export const OURA_OAUTH_LEGACY_TOKEN_URL =
|
||||
"https://api.ouraring.com/oauth/token";
|
||||
|
||||
/** The origin of Oura's current authorization server. */
|
||||
const OURA_MOI_ORIGIN = "https://moi.ouraring.com";
|
||||
|
||||
/**
|
||||
* Authorization-server origins an `iss` callback parameter may name
|
||||
* (RFC 9207). Anything else is a mix-up and refused.
|
||||
*/
|
||||
const OURA_ISSUER_ORIGINS = new Set([
|
||||
OURA_MOI_ORIGIN,
|
||||
"https://cloud.ouraring.com",
|
||||
"https://api.ouraring.com",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Check the `iss` the authorization server sent with the callback. Absent is
|
||||
* fine (older apps never send one); present, it must name one of Oura's own
|
||||
* authorization servers.
|
||||
*/
|
||||
export function isAcceptedOuraIssuer(iss: string | null): boolean {
|
||||
if (iss === null) return true;
|
||||
try {
|
||||
const url = new URL(iss);
|
||||
return url.protocol === "https:" && OURA_ISSUER_ORIGINS.has(url.origin);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** True when the code was issued by the current authorization server. */
|
||||
function issuedByMoi(iss: string | null | undefined): boolean {
|
||||
if (!iss) return false;
|
||||
try {
|
||||
return new URL(iss).origin === OURA_MOI_ORIGIN;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export interface OuraCredentials {
|
||||
clientId: string;
|
||||
@@ -76,12 +134,18 @@ export interface OuraTokenResponse {
|
||||
scope?: string;
|
||||
}
|
||||
|
||||
async function postToken(
|
||||
interface TokenAttempt {
|
||||
status: number;
|
||||
json: unknown;
|
||||
}
|
||||
|
||||
async function requestToken(
|
||||
url: string,
|
||||
params: URLSearchParams,
|
||||
verb: string,
|
||||
): Promise<OuraTokenResponse> {
|
||||
): Promise<TokenAttempt> {
|
||||
const start = performance.now();
|
||||
const res = await safeFetch(OURA_OAUTH_TOKEN_URL, {
|
||||
const res = await safeFetch(url, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
@@ -89,7 +153,7 @@ async function postToken(
|
||||
},
|
||||
body: params.toString(),
|
||||
});
|
||||
const json = await res.json().catch(() => null);
|
||||
const json: unknown = await res.json().catch(() => null);
|
||||
const verdict = classifyOuraResponse(res.status);
|
||||
getEvent()?.addExternalCall({
|
||||
service: "oura",
|
||||
@@ -98,33 +162,91 @@ async function postToken(
|
||||
status: res.status,
|
||||
error: verdict.classification === "success" ? undefined : verdict.reason,
|
||||
});
|
||||
return { status: res.status, json };
|
||||
}
|
||||
|
||||
function upstreamError(json: unknown): string | undefined {
|
||||
if (json === null || typeof json !== "object") return undefined;
|
||||
const error = (json as { error?: unknown }).error;
|
||||
return typeof error === "string" ? error : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the legacy endpoint is worth asking after the live one refused:
|
||||
* a 400, 401 or 404. The live endpoint may not know an older app's client,
|
||||
* code or refresh token at all, and its refusal does not tell "unknown here"
|
||||
* apart from "bad grant" reliably, so every such refusal gets the one retry.
|
||||
* Server errors and rate limits do not: they say nothing about the grant.
|
||||
*/
|
||||
const FALLBACK_STATUSES = new Set([400, 401, 404]);
|
||||
|
||||
function belongsElsewhere(attempt: TokenAttempt): boolean {
|
||||
return FALLBACK_STATUSES.has(attempt.status);
|
||||
}
|
||||
|
||||
function tokenResult(attempt: TokenAttempt, verb: string): OuraTokenResponse {
|
||||
const verdict = classifyOuraResponse(attempt.status);
|
||||
if (verdict.classification !== "success") {
|
||||
throw new OuraApiError({
|
||||
verb,
|
||||
classification: verdict.classification,
|
||||
httpStatus: verdict.httpStatus,
|
||||
reason: verdict.reason,
|
||||
upstreamError: typeof json?.error === "string" ? json.error : undefined,
|
||||
upstreamError: upstreamError(attempt.json),
|
||||
});
|
||||
}
|
||||
// A 2xx with an empty or non-JSON body (captive portal, gateway 200, 204)
|
||||
// leaves `json` null. Casting it straight to the token type handed callers a
|
||||
// null they then dereferenced, surfacing an unclassified TypeError instead of
|
||||
// a handled integration error. Narrow explicitly and classify as transient.
|
||||
if (json === null || typeof json !== "object") {
|
||||
if (attempt.json === null || typeof attempt.json !== "object") {
|
||||
throw new OuraApiError({
|
||||
verb,
|
||||
classification: "transient",
|
||||
httpStatus: res.status,
|
||||
httpStatus: attempt.status,
|
||||
reason: "empty_token_body",
|
||||
});
|
||||
}
|
||||
return json as OuraTokenResponse;
|
||||
return attempt.json as OuraTokenResponse;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ask the live endpoint, then the legacy one when `mayFallBack` allows it and
|
||||
* the live endpoint refused with 400, 401 or 404. The legacy endpoint is
|
||||
* never asked first.
|
||||
*/
|
||||
async function postToken(
|
||||
params: URLSearchParams,
|
||||
verb: string,
|
||||
mayFallBack: boolean,
|
||||
): Promise<OuraTokenResponse> {
|
||||
const live = await requestToken(OURA_OAUTH_TOKEN_URL, params, verb);
|
||||
if (mayFallBack && belongsElsewhere(live)) {
|
||||
const legacy = await requestToken(
|
||||
OURA_OAUTH_LEGACY_TOKEN_URL,
|
||||
params,
|
||||
`${verb}.legacy`,
|
||||
);
|
||||
if (legacy.status >= 200 && legacy.status < 300) {
|
||||
return tokenResult(legacy, verb);
|
||||
}
|
||||
// Neither endpoint took it. The live endpoint's answer is the one that
|
||||
// describes the grant for every current app, so report that one.
|
||||
}
|
||||
return tokenResult(live, verb);
|
||||
}
|
||||
|
||||
/**
|
||||
* Exchange an authorization code. `iss` is the callback's issuer parameter:
|
||||
* a code from the current authorization server goes to the live endpoint
|
||||
* only, since a second attempt elsewhere could never succeed. Without an
|
||||
* issuer (an older app) the legacy endpoint is the fallback after the live
|
||||
* one has refused the code; the live endpoint is always asked first.
|
||||
*/
|
||||
export async function exchangeCode(
|
||||
code: string,
|
||||
creds: OuraCredentials,
|
||||
iss?: string | null,
|
||||
): Promise<OuraTokenResponse> {
|
||||
return postToken(
|
||||
new URLSearchParams({
|
||||
@@ -135,9 +257,15 @@ export async function exchangeCode(
|
||||
client_secret: creds.clientSecret,
|
||||
}),
|
||||
"exchangeCode",
|
||||
!issuedByMoi(iss),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh an access token: live endpoint first, the legacy endpoint when the
|
||||
* live one refuses it (a refresh token of an older app).
|
||||
* A refused refresh does not consume the token, so the retry is safe.
|
||||
*/
|
||||
export async function refreshAccessToken(
|
||||
refreshToken: string,
|
||||
creds: OuraCredentials,
|
||||
@@ -150,6 +278,7 @@ export async function refreshAccessToken(
|
||||
client_secret: creds.clientSecret,
|
||||
}),
|
||||
"refreshAccessToken",
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -31,14 +31,14 @@ export async function checkRateLimit(
|
||||
|
||||
const rows = await prisma.$queryRaw<{ count: number; reset_at: Date }[]>`
|
||||
INSERT INTO rate_limits (key, count, reset_at)
|
||||
VALUES (${key}, 1, NOW() + ${windowInterval}::interval)
|
||||
VALUES (${key}, 1, (NOW() AT TIME ZONE 'UTC') + ${windowInterval}::interval)
|
||||
ON CONFLICT (key) DO UPDATE SET
|
||||
count = CASE
|
||||
WHEN rate_limits.reset_at < NOW() THEN 1
|
||||
WHEN rate_limits.reset_at < (NOW() AT TIME ZONE 'UTC') THEN 1
|
||||
ELSE rate_limits.count + 1
|
||||
END,
|
||||
reset_at = CASE
|
||||
WHEN rate_limits.reset_at < NOW() THEN NOW() + ${windowInterval}::interval
|
||||
WHEN rate_limits.reset_at < (NOW() AT TIME ZONE 'UTC') THEN (NOW() AT TIME ZONE 'UTC') + ${windowInterval}::interval
|
||||
ELSE rate_limits.reset_at
|
||||
END
|
||||
RETURNING count, reset_at
|
||||
@@ -80,7 +80,7 @@ export async function refundRateLimit(key: string): Promise<void> {
|
||||
await prisma.$executeRaw`
|
||||
UPDATE rate_limits
|
||||
SET count = GREATEST(count - 1, 0)
|
||||
WHERE key = ${key} AND reset_at >= NOW()
|
||||
WHERE key = ${key} AND reset_at >= (NOW() AT TIME ZONE 'UTC')
|
||||
`;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user