diff --git a/src/lib/ai/__tests__/codex-oauth.test.ts b/src/lib/ai/__tests__/codex-oauth.test.ts index 1092e26af..dbbc695cf 100644 --- a/src/lib/ai/__tests__/codex-oauth.test.ts +++ b/src/lib/ai/__tests__/codex-oauth.test.ts @@ -225,6 +225,23 @@ describe("codex-oauth", () => { expect(err.oauthError).toBeNull(); }); + it("does not read a bare 403 as a dead sign-in", async () => { + // An edge challenge in front of the issuer answers a valid grant so. + failWith(403, "challenge"); + const err = await refreshError(); + expect(err.credentialExpired).toBe(false); + expect(err.httpStatus).toBe(403); + }); + + it("does not read refresh_token_reused as a dead sign-in", async () => { + // The loser of two refreshes of one token hears this; the winner's + // rotated token is alive. + failWith(400, JSON.stringify({ error: "refresh_token_reused" })); + const err = await refreshError(); + expect(err.credentialExpired).toBe(false); + expect(err.oauthError).toBe("refresh_token_reused"); + }); + it("never quotes the response body in the message", async () => { failWith( 401, diff --git a/src/lib/ai/codex-oauth.ts b/src/lib/ai/codex-oauth.ts index 6d389160c..bcfd2c630 100644 --- a/src/lib/ai/codex-oauth.ts +++ b/src/lib/ai/codex-oauth.ts @@ -296,14 +296,18 @@ export async function refreshDeviceTokens( /** * OAuth error codes on the refresh grant that mean the stored refresh token - * is dead for good: revoked, expired, already rotated, or issued to another - * client. Retrying cannot help; only signing in again can. + * is dead for good: revoked, expired, or issued to another client. Retrying + * cannot help; only signing in again can. + * + * `refresh_token_reused` is deliberately not among them. It is what the loser + * of two refreshes of one token hears while the winner's rotated token is + * perfectly alive, so it says nothing about the sign-in: the refresh re-reads + * the row and carries on with the newer token (`refreshCodexCredential`). */ const DEAD_REFRESH_TOKEN_CODES: ReadonlySet = new Set([ "invalid_grant", "invalid_refresh_token", "refresh_token_expired", - "refresh_token_reused", "refresh_token_invalidated", "invalid_token", ]); @@ -333,9 +337,12 @@ export class CodexRefreshError extends Error { readonly oauthError: string | null; constructor(status: number, oauthError: string | null) { + // A bare 403 is not proof: an edge challenge in front of the issuer + // answers 403 to a perfectly valid grant. It keeps its own status and + // with it the ledger's timed auth cooldown; only a 401 or the issuer's + // own dead-grant code parks the sign-in until the person reconnects. const credentialExpired = status === 401 || - status === 403 || (oauthError !== null && DEAD_REFRESH_TOKEN_CODES.has(oauthError)); super( credentialExpired @@ -345,7 +352,7 @@ export class CodexRefreshError extends Error { this.name = "CodexRefreshError"; this.credentialExpired = credentialExpired; this.oauthError = oauthError; - this.httpStatus = credentialExpired && status !== 403 ? 401 : status; + this.httpStatus = credentialExpired ? 401 : status; } } diff --git a/src/lib/ai/provider.ts b/src/lib/ai/provider.ts index e05758bed..2a5e24b1e 100644 --- a/src/lib/ai/provider.ts +++ b/src/lib/ai/provider.ts @@ -14,6 +14,7 @@ import { decryptCodexCreds, encryptAdminCodexCreds, decryptAdminCodexCreds, + type CodexCreds, } from "./codex-oauth"; import { isPublicUrl } from "@/lib/validations/notifications"; import { isLocalAiHostAllowed } from "./local-host-allowlist"; @@ -31,6 +32,11 @@ import { } from "@/lib/sharing/provider-work-authority"; import { memoizePerRequest } from "@/lib/request-cache"; import { emitSignal } from "@/lib/logging/signal"; +import { + acquireProviderTokenRefreshLock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, +} from "@/lib/integrations/oauth-refresh"; +import type { Prisma } from "@/generated/prisma/client"; import type { AiModality } from "@/lib/ai/capabilities/types"; import type { ProviderEntryPresence, @@ -62,72 +68,253 @@ function oauthErrorOf(error: unknown): string | null { return typeof code === "string" ? code : null; } +type CodexCredentialOwner = "codex" | "admin-codex"; + /** - * Park a Codex connection whose sign-in the issuer has refused for good. - * - * Retrying a dead refresh token cannot succeed; until this existed every - * generation of every card of every run asked the issuer again, failed with - * the same 401, and logged it. Parking flips the connection to the existing - * `expired` state — the one Settings already renders as "reconnect ChatGPT" — - * so the chain, the presence probe and with it every AI capability stop - * counting it, and nothing is sent until the person signs in again (the - * device-code flow writes `connected`). - * - * Compare-and-set on the refresh token that was refused: a reconnect that - * landed in between wrote a different one and is left alone, and of several - * workers failing at once exactly one flips the row, so the transition is - * signalled once and never again per run. + * One Codex credential row as a refresh sees it: the user's own connection or + * the operator's central one. `writeRotated` and `park` are compare-and-set on + * the refresh token this refresh spent, so a sign-in that landed meanwhile (the + * device-code flow writes without the refresh lock) is never overwritten. */ -async function parkUserCodex( - userId: string, - refusedRefreshEncrypted: string, - error: unknown, -): Promise { - try { - const parked = await prisma.user.updateMany({ - where: { - id: userId, - codexConnectionStatus: "connected", - codexRefreshTokenEncrypted: refusedRefreshEncrypted, - }, - data: { codexConnectionStatus: "expired" }, - }); - if (parked.count === 1) { - emitSignal({ - action: "ai.credential.expired", - level: "warn", - meta: { provider: "codex", oauth_error: oauthErrorOf(error) }, - }); - } - } catch { - // Best effort: an unparked connection is retried exactly as before. - } +interface CodexCredentialStore { + owner: CodexCredentialOwner; + lockId: string; + read(tx: Prisma.TransactionClient): Promise<{ + status: string | null; + refreshEncrypted: string | null; + creds: CodexCreds | null; + } | null>; + writeRotated( + tx: Prisma.TransactionClient, + spentRefreshEncrypted: string, + fresh: CodexCreds, + ): Promise; + park( + tx: Prisma.TransactionClient, + spentRefreshEncrypted: string, + ): Promise; } -/** The operator's central Codex, parked the same way. */ -async function parkCentralCodex( - refusedRefreshEncrypted: string, - error: unknown, -): Promise { - try { - const parked = await prisma.appSettings.updateMany({ +function userCodexStore(userId: string): CodexCredentialStore { + return { + owner: "codex", + lockId: userId, + async read(tx) { + const row = await tx.user.findUnique({ + where: { id: userId }, + select: { + codexConnectionStatus: true, + codexAccessTokenEncrypted: true, + codexRefreshTokenEncrypted: true, + }, + }); + if (!row) return null; + return { + status: row.codexConnectionStatus, + refreshEncrypted: row.codexRefreshTokenEncrypted, + creds: + row.codexAccessTokenEncrypted && row.codexRefreshTokenEncrypted + ? decryptCodexCreds({ + accessEncrypted: row.codexAccessTokenEncrypted, + refreshEncrypted: row.codexRefreshTokenEncrypted, + }) + : null, + }; + }, + async writeRotated(tx, spent, fresh) { + const enc = encryptCodexCreds(fresh); + const written = await tx.user.updateMany({ + where: { id: userId, codexRefreshTokenEncrypted: spent }, + data: { + codexAccessTokenEncrypted: enc.accessEncrypted, + codexRefreshTokenEncrypted: enc.refreshEncrypted, + codexTokenExpiresAt: fresh.expiresAt, + codexConnectionStatus: "connected", + }, + }); + return written.count; + }, + async park(tx, spent) { + const parked = await tx.user.updateMany({ + where: { + id: userId, + codexConnectionStatus: "connected", + codexRefreshTokenEncrypted: spent, + }, + data: { codexConnectionStatus: "expired" }, + }); + return parked.count; + }, + }; +} + +const centralCodexStore: CodexCredentialStore = { + owner: "admin-codex", + lockId: "singleton", + async read(tx) { + const row = await tx.appSettings.findUnique({ + where: { id: "singleton" }, + select: { + adminCodexConnectionStatus: true, + adminCodexAccessTokenEncrypted: true, + adminCodexRefreshTokenEncrypted: true, + adminCodexAccountIdEncrypted: true, + adminCodexTokenExpiresAt: true, + }, + }); + if (!row) return null; + return { + status: row.adminCodexConnectionStatus, + refreshEncrypted: row.adminCodexRefreshTokenEncrypted, + creds: + row.adminCodexAccessTokenEncrypted && + row.adminCodexRefreshTokenEncrypted && + row.adminCodexAccountIdEncrypted + ? decryptAdminCodexCreds({ + accessEncrypted: row.adminCodexAccessTokenEncrypted, + refreshEncrypted: row.adminCodexRefreshTokenEncrypted, + accountIdEncrypted: row.adminCodexAccountIdEncrypted, + expiresAt: row.adminCodexTokenExpiresAt, + }) + : null, + }; + }, + async writeRotated(tx, spent, fresh) { + const enc = encryptAdminCodexCreds(fresh); + const written = await tx.appSettings.updateMany({ + where: { id: "singleton", adminCodexRefreshTokenEncrypted: spent }, + data: { + adminCodexAccessTokenEncrypted: enc.accessEncrypted, + adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, + adminCodexAccountIdEncrypted: enc.accountIdEncrypted, + adminCodexTokenExpiresAt: enc.expiresAt, + adminCodexConnectionStatus: "connected", + }, + }); + return written.count; + }, + async park(tx, spent) { + const parked = await tx.appSettings.updateMany({ where: { id: "singleton", adminCodexConnectionStatus: "connected", - adminCodexRefreshTokenEncrypted: refusedRefreshEncrypted, + adminCodexRefreshTokenEncrypted: spent, }, data: { adminCodexConnectionStatus: "expired" }, }); - if (parked.count === 1) { - emitSignal({ - action: "ai.credential.expired", - level: "warn", - meta: { provider: "admin-codex", oauth_error: oauthErrorOf(error) }, - }); - } - } catch { - // Best effort, as above. + return parked.count; + }, +}; + +/** Refreshes in flight in this process, keyed on credential and spent token. */ +const codexRefreshesInFlight = new Map>(); + +/** + * Refresh one Codex credential, at most once per spent token. + * + * The issuer rotates the refresh token on every refresh and honours each one + * once. Two callers that refreshed the same token used to race: the loser got + * `refresh_token_reused`, read it as a dead sign-in and parked the connection + * while the winner's fresh tokens landed beside an `expired` status. So: + * + * - callers in one process share one refresh (single flight), and callers + * in different processes queue on the per-credential advisory lock the + * provider syncs use (`acquireProviderTokenRefreshLock`); + * - inside the lock the row is read again, and a token a peer already + * rotated is used as it is, without asking the issuer; + * - the rotated tokens are written compare-and-set on the spent token, and + * the write sets `connected`; + * - after a refusal the row is read again too, and only a refusal of the + * token still stored, with the issuer's dead-grant answer, parks it + * (`expired`, the state Settings renders as "reconnect ChatGPT"). The + * transition is signalled once: only the lock holder can make it. + * + * Parking stops the chain, the presence probe and every AI capability from + * counting the connection until the person signs in again, which is what + * stopped every run from asking the issuer about the same dead token. + */ +function refreshCodexCredential( + store: CodexCredentialStore, + spentRefreshEncrypted: string, +): Promise { + const key = `${store.owner}:${store.lockId}:${spentRefreshEncrypted}`; + const inFlight = codexRefreshesInFlight.get(key); + if (inFlight) return inFlight; + const run = runCodexRefresh(store, spentRefreshEncrypted).finally(() => { + codexRefreshesInFlight.delete(key); + }); + codexRefreshesInFlight.set(key, run); + return run; +} + +async function runCodexRefresh( + store: CodexCredentialStore, + spent: string, +): Promise { + const outcome = await prisma.$transaction( + async ( + tx, + ): Promise< + | { fresh: CodexCreds } + | { refused: unknown; parked: boolean } + | { failed: unknown } + > => { + await acquireProviderTokenRefreshLock(tx, store.owner, store.lockId); + const row = await store.read(tx); + if (!row || row.status !== "connected" || !row.refreshEncrypted) { + return { + refused: codexReconnectRequired("no refresh token"), + parked: false, + }; + } + if (row.refreshEncrypted !== spent && row.creds) { + // A peer rotated while this caller waited: its token is the live one. + return { fresh: row.creds }; + } + if (!row.creds) { + const refused = codexReconnectRequired("token storage corrupt"); + return { + refused, + parked: (await store.park(tx, row.refreshEncrypted)) === 1, + }; + } + let fresh: CodexCreds; + try { + fresh = await refreshDeviceTokens(row.creds.refreshToken); + } catch (error) { + const now = await store.read(tx); + if (now?.refreshEncrypted !== row.refreshEncrypted && now?.creds) { + // A sign-in replaced the token while the issuer refused the old one. + return { fresh: now.creds }; + } + if (!isDeadCodexSignIn(error)) return { failed: error }; + return { + refused: error, + parked: (await store.park(tx, row.refreshEncrypted)) === 1, + }; + } + if ((await store.writeRotated(tx, row.refreshEncrypted, fresh)) === 0) { + // A sign-in replaced the row meanwhile; it wins over this refresh. + const now = await store.read(tx); + if (now?.creds) return { fresh: now.creds }; + } + return { fresh }; + }, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, + ); + if ("fresh" in outcome) return outcome.fresh; + if ("failed" in outcome) throw outcome.failed; + if (outcome.parked) { + emitSignal({ + action: "ai.credential.expired", + level: "warn", + meta: { + provider: store.owner, + oauth_error: oauthErrorOf(outcome.refused), + }, + }); } + throw outcome.refused; } class NoProvider implements AIProvider { @@ -384,27 +571,16 @@ async function resolveCodexProvider( // Proactive refresh if the access token is within 5 min of expiry. if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) { try { - const fresh = await refreshDeviceTokens(stored.refreshToken); - active = fresh; - - const enc = encryptCodexCreds(fresh); - await prisma.user.update({ - where: { id: userId }, - data: { - codexAccessTokenEncrypted: enc.accessEncrypted, - codexRefreshTokenEncrypted: enc.refreshEncrypted, - codexTokenExpiresAt: fresh.expiresAt, - }, - }); + active = await refreshCodexCredential( + userCodexStore(userId), + user.codexRefreshTokenEncrypted, + ); } catch (error) { - // A refused grant parks the connection and drops it from this chain: - // calling Codex with the expired access token could only 401 into a - // second refusal. Anything else falls through — CodexClient will - // trigger an on-401 refresh and persist there. - if (isDeadCodexSignIn(error)) { - await parkUserCodex(userId, user.codexRefreshTokenEncrypted, error); - return null; - } + // A refused grant drops the connection from this chain (the refresh + // parked it): calling Codex with the expired access token could only + // 401 into a second refusal. Anything else falls through — CodexClient + // will trigger an on-401 refresh. + if (isDeadCodexSignIn(error)) return null; } } @@ -412,54 +588,17 @@ async function resolveCodexProvider( accessToken: active.accessToken, accountId: active.accountId, onTokenRefresh: async () => { - const freshUser = await prisma.user.findUnique({ + const current = await prisma.user.findUnique({ where: { id: userId }, - select: { - codexAccessTokenEncrypted: true, - codexRefreshTokenEncrypted: true, - }, + select: { codexRefreshTokenEncrypted: true }, }); - if ( - !freshUser?.codexAccessTokenEncrypted || - !freshUser.codexRefreshTokenEncrypted - ) { + if (!current?.codexRefreshTokenEncrypted) { throw codexReconnectRequired("no refresh token"); } - const decoded = decryptCodexCreds({ - accessEncrypted: freshUser.codexAccessTokenEncrypted, - refreshEncrypted: freshUser.codexRefreshTokenEncrypted, - }); - if (!decoded) { - const error = codexReconnectRequired("token storage corrupt"); - await parkUserCodex( - userId, - freshUser.codexRefreshTokenEncrypted, - error, - ); - throw error; - } - let fresh; - try { - fresh = await refreshDeviceTokens(decoded.refreshToken); - } catch (error) { - if (isDeadCodexSignIn(error)) { - await parkUserCodex( - userId, - freshUser.codexRefreshTokenEncrypted, - error, - ); - } - throw error; - } - const enc = encryptCodexCreds(fresh); - await prisma.user.update({ - where: { id: userId }, - data: { - codexAccessTokenEncrypted: enc.accessEncrypted, - codexRefreshTokenEncrypted: enc.refreshEncrypted, - codexTokenExpiresAt: fresh.expiresAt, - }, - }); + const fresh = await refreshCodexCredential( + userCodexStore(userId), + current.codexRefreshTokenEncrypted, + ); return { accessToken: fresh.accessToken, accountId: fresh.accountId }; }, }); @@ -521,26 +660,15 @@ async function resolveAdminCodexProvider(): Promise { if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) { try { - const fresh = await refreshDeviceTokens(stored.refreshToken); - active = fresh; - const enc = encryptAdminCodexCreds(fresh); - await prisma.appSettings.update({ - where: { id: "singleton" }, - data: { - adminCodexAccessTokenEncrypted: enc.accessEncrypted, - adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, - adminCodexAccountIdEncrypted: enc.accountIdEncrypted, - adminCodexTokenExpiresAt: enc.expiresAt, - }, - }); + active = await refreshCodexCredential( + centralCodexStore, + settings.adminCodexRefreshTokenEncrypted, + ); } catch (error) { - // A refused grant parks the central connection (see `parkUserCodex`); - // anything else falls through — CodexClient triggers an on-401 refresh - // and persists via the callback below. - if (isDeadCodexSignIn(error)) { - await parkCentralCodex(settings.adminCodexRefreshTokenEncrypted, error); - return null; - } + // A refused grant parked the central connection (see + // `refreshCodexCredential`); anything else falls through — CodexClient + // triggers an on-401 refresh via the callback below. + if (isDeadCodexSignIn(error)) return null; } } @@ -548,52 +676,17 @@ async function resolveAdminCodexProvider(): Promise { accessToken: active.accessToken, accountId: active.accountId, onTokenRefresh: async () => { - const fresh = await prisma.appSettings.findUnique({ + const current = await prisma.appSettings.findUnique({ where: { id: "singleton" }, - select: { - adminCodexAccessTokenEncrypted: true, - adminCodexRefreshTokenEncrypted: true, - adminCodexAccountIdEncrypted: true, - adminCodexTokenExpiresAt: true, - }, + select: { adminCodexRefreshTokenEncrypted: true }, }); - if ( - !fresh?.adminCodexAccessTokenEncrypted || - !fresh.adminCodexRefreshTokenEncrypted || - !fresh.adminCodexAccountIdEncrypted - ) { + if (!current?.adminCodexRefreshTokenEncrypted) { throw codexReconnectRequired("no central refresh token"); } - const decoded = decryptAdminCodexCreds({ - accessEncrypted: fresh.adminCodexAccessTokenEncrypted, - refreshEncrypted: fresh.adminCodexRefreshTokenEncrypted, - accountIdEncrypted: fresh.adminCodexAccountIdEncrypted, - expiresAt: fresh.adminCodexTokenExpiresAt, - }); - if (!decoded) { - const error = codexReconnectRequired("central token storage corrupt"); - await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); - throw error; - } - let refreshed; - try { - refreshed = await refreshDeviceTokens(decoded.refreshToken); - } catch (error) { - if (isDeadCodexSignIn(error)) { - await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); - } - throw error; - } - const enc = encryptAdminCodexCreds(refreshed); - await prisma.appSettings.update({ - where: { id: "singleton" }, - data: { - adminCodexAccessTokenEncrypted: enc.accessEncrypted, - adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, - adminCodexAccountIdEncrypted: enc.accountIdEncrypted, - adminCodexTokenExpiresAt: enc.expiresAt, - }, - }); + const refreshed = await refreshCodexCredential( + centralCodexStore, + current.adminCodexRefreshTokenEncrypted, + ); return { accessToken: refreshed.accessToken, accountId: refreshed.accountId, diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts index ca0da45e3..452caec8f 100644 --- a/src/lib/integrations/oauth-refresh.ts +++ b/src/lib/integrations/oauth-refresh.ts @@ -61,7 +61,8 @@ export async function persistRotatedToken( return opts.readPeerAccessToken(); } -export type SerializedRefreshProvider = "whoop" | "withings" | "oura"; +export type SerializedRefreshProvider = + "whoop" | "withings" | "oura" | "codex" | "admin-codex"; /** * Provider refreshes hold a database transaction open around one bounded diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 3c3d007ff..3bc085fd0 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -190,7 +190,9 @@ export const QUEUE_RUNTIME: Readonly> = { "daily-briefing": short("A light cohort scan every fifteen minutes."), // ── Rollups ──────────────────────────────────────────────────────── - "rollup-recompute": short("One bucket."), + "rollup-recompute": short( + "One type at one granularity over one bucket or a run of adjacent buckets, inside the five-year fold window.", + ), "rollup-full-backfill": { runtime: "long", expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 0a9eae21b..a5277687f 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -362,8 +362,19 @@ export async function recomputeBucketsForMeasurement( ); } +/** + * UTC wall-clock literals for a `timestamp[]` parameter. `measured_at` and + * `bucket_start` are naive UTC timestamps, so the instants they are compared + * with must be too: a `timestamptz[]` parameter is converted through the + * session time zone, and in any zone but UTC neither the day filter nor the + * keyed delete matched a single day, so a refold left its buckets stale. + */ +function utcStamps(dates: readonly Date[]): string[] { + return dates.map((d) => d.toISOString().replace("Z", "")); +} + /** Keys one set-based DAY refold carries; bounds the statement's arrays. */ -const KEYED_REFOLD_CHUNK = 2_000; +const KEYED_REFOLD_CHUNK = 500; /** * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write @@ -381,7 +392,17 @@ export async function recomputeBucketsForKeys( userId: string, rows: Array<{ type: MeasurementType; measuredAt: Date }>, ): Promise { - const keys = collapseToTypeDayKeys(rows); + // Sorted on (type, day), the order every upsert of the tier takes its row + // locks in (`persistRollupRows` sorts the same way), and chunked small, so + // a keyed refold racing the full fold or another refold holds a short, + // deterministic slice and waits rather than interleaving lock orders. + const keys = collapseToTypeDayKeys(rows).sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.measuredAt.getTime() - b.measuredAt.getTime(), + ); for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) { await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK)); } @@ -441,11 +462,11 @@ async function refoldDayBuckets( AND granularity = 'DAY'::"measurement_rollup_granularity" AND (type, bucket_start) IN ( SELECT k.t::"measurement_type", k.b - FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b) + FROM unnest($2::text[], $3::timestamp[]) AS k(t, b) )`, userId, keys.map((k) => k.type), - keys.map((k) => k.measuredAt.toISOString()), + utcStamps(keys.map((k) => k.measuredAt)), ); for (let i = 0; i < ordered.length; i += 500) { const { sql, params } = buildRollupUpsert( @@ -642,9 +663,9 @@ async function runRollupAggregate( if (input.types && input.types.length === 0) return []; // Bound as the fourth parameter, never spliced. const dayFilter = input.days - ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])` : ""; - const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : []; + const dayParams = input.days ? [utcStamps(input.days)] : []; // Postgres rejects `date_trunc($1::text, m."measured_at")` in a // GROUP BY because the planner sees the parameterised expression @@ -857,7 +878,7 @@ async function applyHourlyMeanDay( AND m."deleted_at" IS NULL ${ input.days - ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])` : "" } GROUP BY m."type", m."source", date_trunc('hour', m."measured_at") @@ -869,7 +890,7 @@ async function applyHourlyMeanDay( input.userId, input.from, input.to, - ...(input.days ? [input.days.map((d) => d.toISOString())] : []), + ...(input.days ? [utcStamps(input.days)] : []), ); const byKey = new Map( means.map((m) => [ diff --git a/tests/integration/codex-credential-park.test.ts b/tests/integration/codex-credential-park.test.ts index 95ea57391..811fefd0c 100644 --- a/tests/integration/codex-credential-park.test.ts +++ b/tests/integration/codex-credential-park.test.ts @@ -261,3 +261,61 @@ describe("a refused Codex sign-in", () => { expect(vi.mocked(refreshDeviceTokens)).toHaveBeenCalledTimes(1); }); }); + +describe("two refreshes of one Codex token", () => { + async function codexStatus(userId: string) { + return ( + await prisma.user.findUniqueOrThrow({ + where: { id: userId }, + select: { codexConnectionStatus: true }, + }) + ).codexConnectionStatus; + } + + it("share one refresh: the slower caller never parks the winner's sign-in", async () => { + const user = await seedCodexUser(); + // The issuer honours each refresh token once. The first caller's refresh + // takes a moment; a second refresh of the same token is refused as reused. + let asked = 0; + vi.mocked(refreshDeviceTokens).mockImplementation(async () => { + asked += 1; + if (asked > 1) throw new CodexRefreshError(400, "refresh_token_reused"); + await new Promise((resolve) => setTimeout(resolve, 200)); + return creds("rotated", 10 * HOUR_MS); + }); + + const [first, second] = await Promise.all([ + resolveProviderChain(user.id), + resolveProviderChain(user.id), + ]); + + expect(first.map((p) => p.providerType)).toContain("codex"); + expect(second.map((p) => p.providerType)).toContain("codex"); + expect(asked).toBe(1); + expect(await codexStatus(user.id)).toBe("connected"); + expect(parkSignals("codex")).toHaveLength(0); + }); + + it("carries on with the token a peer rotated while the issuer refused the old one", async () => { + const user = await seedCodexUser(); + // Another process won the race: by the time the issuer answers this + // caller "reused", the peer's rotated tokens are in the row. + vi.mocked(refreshDeviceTokens).mockImplementation(async () => { + const peer = encryptCodexCreds(creds("peer", 10 * HOUR_MS)); + await prisma.user.update({ + where: { id: user.id }, + data: { + codexAccessTokenEncrypted: peer.accessEncrypted, + codexRefreshTokenEncrypted: peer.refreshEncrypted, + }, + }); + throw new CodexRefreshError(400, "refresh_token_reused"); + }); + + const chain = await resolveProviderChain(user.id); + + expect(chain.map((p) => p.providerType)).toContain("codex"); + expect(await codexStatus(user.id)).toBe("connected"); + expect(parkSignals("codex")).toHaveLength(0); + }); +}); diff --git a/tests/integration/rollup-keyed-refold-session-tz.test.ts b/tests/integration/rollup-keyed-refold-session-tz.test.ts new file mode 100644 index 000000000..30e88bb3c --- /dev/null +++ b/tests/integration/rollup-keyed-refold-session-tz.test.ts @@ -0,0 +1,76 @@ +/** + * The keyed DAY refold does not depend on the session time zone. + * + * `measured_at` and `bucket_start` are naive UTC timestamps. The refold's day + * filter and its delete compared them with a `timestamptz[]` parameter, which + * Postgres converts through the session time zone: outside UTC neither matched + * a single day, so the refold silently left every touched bucket stale. The + * app pins its sessions to UTC, so this runs the pool in another zone on + * purpose. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db-session", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { prisma } from "@/lib/db"; +import { + recomputeBucketsForKeys, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("keyed DAY refold in a non-UTC session", () => { + it("folds a new reading into its day", async () => { + const [{ zone }] = await prisma.$queryRaw>` + SELECT current_setting('TimeZone') AS zone + `; + expect(zone).toBe("Pacific/Auckland"); + + const user = await getPrismaClient().user.create({ + data: { username: "tz-refold", email: "tz-refold@example.test" }, + }); + const day = new Date("2026-05-10T00:00:00.000Z"); + await getPrismaClient().measurement.createMany({ + data: [7, 19].map((hour, i) => ({ + userId: user.id, + type: "WEIGHT" as const, + value: 80 + i, + unit: "kg", + measuredAt: new Date(day.getTime() + hour * 3_600_000), + })), + }); + await recomputeBucketsForMeasurement(user.id, "WEIGHT", day); + // A third reading lands; the refold has to fold it in. + await getPrismaClient().measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 82, + unit: "kg", + measuredAt: new Date(day.getTime() + 21 * 3_600_000), + }, + }); + + await recomputeBucketsForKeys(user.id, [ + { type: "WEIGHT", measuredAt: new Date(day.getTime() + 7 * 3_600_000) }, + ]); + + const rows = await getPrismaClient().measurementRollup.findMany({ + where: { userId: user.id, granularity: "DAY" }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]!.bucketStart.toISOString()).toBe(day.toISOString()); + expect(rows[0]!.count).toBe(3); + }); +});