mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-14 14:40:52 +00:00
fix(db): read the clock as UTC in raw SQL and prove it on a Berlin database
Every raw-SQL clock read against a timestamp column without a zone now writes NOW() AT TIME ZONE 'UTC', so passkey and login challenges, rate limits, provider pauses, coach usage stamps, tombstone and measurement windows no longer pass through the session time zone even where the per-connection pin cannot apply. The regression-window bounds stay naive UTC timestamps instead of being turned into timestamptz. A structural guard keeps bare NOW() / CURRENT_TIMESTAMP out of raw SQL outside a short list of timestamptz targets. The integration test sets the database default to Europe/Berlin and drives challenge claims, session expiry, the native sign-in handoff freshness check and the rate-limit window through the app's client, plus the challenge claim through an unpinned client. Self-hosting docs describe the pin for an existing or shared Postgres.
This commit is contained in:
1 parent
cefc8c15c1
commit
f65d25b0d0
22 files changed
+469
-66
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);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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")
|
||||
|
||||
@@ -321,12 +321,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 +453,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 +472,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}
|
||||
`;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
+22
-6
@@ -1,6 +1,6 @@
|
||||
import { Prisma, PrismaClient } from "@/generated/prisma/client";
|
||||
import { PrismaPg } from "@prisma/adapter-pg";
|
||||
import { Pool } from "pg";
|
||||
import { Pool, type PoolConfig } from "pg";
|
||||
import {
|
||||
pinTimeZoneOnConnect,
|
||||
sessionOptionsDisabled,
|
||||
@@ -144,16 +144,32 @@ export function buildSessionOptions(): string | undefined {
|
||||
return `${timeouts}-c work_mem=${getWorkMem()} ${TIME_ZONE_OPTION}`;
|
||||
}
|
||||
|
||||
function createPrismaClient() {
|
||||
const pool = new Pool({
|
||||
export function prismaPoolConfig(): PoolConfig {
|
||||
return {
|
||||
connectionString: process.env.DATABASE_URL!,
|
||||
max: getPrismaPoolMax(),
|
||||
connectionTimeoutMillis: getPoolConnectionTimeoutMs(),
|
||||
options: buildSessionOptions(),
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 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);
|
||||
const adapter = new PrismaPg(pool, { disposeExternalPool: true });
|
||||
return new PrismaClient({ adapter });
|
||||
return new PrismaPg(pool);
|
||||
}
|
||||
|
||||
function createPrismaClient() {
|
||||
return new PrismaClient({ adapter: createPrismaAdapter() });
|
||||
}
|
||||
|
||||
// One client per process, in production too. A production build gives every
|
||||
|
||||
@@ -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')`;
|
||||
|
||||
@@ -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
|
||||
`,
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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}
|
||||
`;
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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