fix(db): check callers against the adapter that actually runs

The self-hosted client was typed as a full Supabase client, so an old scoring poll called .single() on a plain promise and failed. The new DataClient only lists methods the adapter implements.

Co-Authored-By: Grok (agent loop)
This commit is contained in:
jesse-ux
2026-10-15 09:21:14 +08:00
co-authored by Grok (agent loop)
parent 9274491681
commit b184448871
19 changed files with 366 additions and 47 deletions
+1 -1
View File
@@ -56,7 +56,7 @@ export async function GET() {
.order("updated_at", { ascending: false });
if (error) throw error;
const profiles = (data ?? []).map((row) => toClientProfile(row as Record<string, unknown>));
const profiles = (data ?? []).map((row: unknown) => toClientProfile(row as Record<string, unknown>));
return NextResponse.json({
profiles,
limit: MAX_SAVED_SUBJECTS,
@@ -1,4 +1,4 @@
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { z } from "zod";
import {
parseRectificationQuestionnaire,
@@ -34,7 +34,7 @@ export type JourneyLoadClient = {
from(table: string): { select(columns: string): JourneyLoadQuery };
};
export function createJourneyLoadClient(supabase: SupabaseClient): JourneyLoadClient {
export function createJourneyLoadClient(supabase: DataClient): JourneyLoadClient {
return {
from(table) {
return {
+2 -2
View File
@@ -1,6 +1,6 @@
import "server-only";
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import type {
BirthTimeJourneyStore,
} from "./birth-time-journey-service.ts";
@@ -24,7 +24,7 @@ import { createDynamicScoringJobStore } from "./birth-time-dynamic-scoring-job-s
export { BirthTimeJourneyStoreError } from "./birth-time-journey-turn-persistence.ts";
export function createSupabaseBirthTimeJourneyStore(
supabase: SupabaseClient,
supabase: DataClient,
now: () => Date = () => new Date(),
): BirthTimeJourneyStore {
const loadClient = createJourneyLoadClient(supabase);
@@ -1,4 +1,4 @@
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { z } from "zod";
import type { JourneySnapshot } from "./birth-time-journey.ts";
import type { LegacyStoredRectificationCase, StoredRectificationCase } from "./birth-time-journey-service.ts";
@@ -56,7 +56,7 @@ function canonicalAskedDomains(input: readonly EvidenceDomain[]): readonly Evide
}
export function createJourneyTurnPersistence(
supabase: SupabaseClient,
supabase: DataClient,
loadCase: (userId: string, caseId: string) => Promise<StoredRectificationCase | null>,
) {
return {
@@ -1,4 +1,4 @@
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { z } from "zod";
import { BirthTimeScoringJobError } from "./birth-time-scoring-job.ts";
import type {
@@ -58,7 +58,7 @@ async function requireReloaded(
}
export function createSupabaseScoringJobStore(
supabase: SupabaseClient,
supabase: DataClient,
loadCase: (userId: string, caseId: string) => Promise<StoredRectificationCase | null>,
): ScoringStoreMethods {
return {
@@ -91,18 +91,17 @@ export function createSupabaseScoringJobStore(
},
async claimScoringJob(identity) {
const { data, error } = await supabase
.rpc("claim_birth_time_scoring_job", {
p_user_id: identity.userId,
p_case_id: identity.caseId,
p_job_id: identity.jobId,
p_evidence_fingerprint: identity.evidenceFingerprint,
p_algorithm_version: identity.algorithmVersion,
p_now: identity.now,
})
.single();
const { data, error } = await supabase.rpc("claim_birth_time_scoring_job", {
p_user_id: identity.userId,
p_case_id: identity.caseId,
p_job_id: identity.jobId,
p_evidence_fingerprint: identity.evidenceFingerprint,
p_algorithm_version: identity.algorithmVersion,
p_now: identity.now,
});
if (error) throw mapScoringRpcError(error.message, error.code);
const parsed = claimSchema.safeParse(data);
const row = Array.isArray(data) ? data[0] ?? null : data;
const parsed = claimSchema.safeParse(row);
if (!parsed.success) throw new BirthTimeJourneyStoreError("load_case");
return {
kind: parsed.data.claim_state,
+46
View File
@@ -0,0 +1,46 @@
export type DataQueryError = Readonly<{ message: string; code?: string }>;
// eslint-disable-next-line @typescript-eslint/no-explicit-any -- phase 1: equals SupabaseClient<any>; phase 2 narrows per table
type UntypedData = any;
export type DataQueryResult = Readonly<{
data: UntypedData;
error: DataQueryError | null;
count?: number | null;
}>;
export interface DataQueryBuilder extends PromiseLike<DataQueryResult> {
select(columns?: string): DataQueryBuilder;
insert(value: Record<string, unknown> | readonly Record<string, unknown>[]): DataQueryBuilder;
upsert(
value: Record<string, unknown> | readonly Record<string, unknown>[],
options?: { onConflict?: string },
): DataQueryBuilder;
update(values: Record<string, unknown>): DataQueryBuilder;
delete(options?: { count?: string }): DataQueryBuilder;
eq(column: string, value: unknown): DataQueryBuilder;
neq(column: string, value: unknown): DataQueryBuilder;
gt(column: string, value: unknown): DataQueryBuilder;
lte(column: string, value: unknown): DataQueryBuilder;
like(column: string, value: unknown): DataQueryBuilder;
in(column: string, value: readonly unknown[]): DataQueryBuilder;
is(column: string, value: unknown): DataQueryBuilder;
not(column: string, operator: string, value: unknown): DataQueryBuilder;
or(expression: string): DataQueryBuilder;
order(column: string, options?: { ascending?: boolean }): DataQueryBuilder;
limit(value: number): DataQueryBuilder;
abortSignal(signal: AbortSignal): DataQueryBuilder;
single(): Promise<DataQueryResult>;
maybeSingle(): Promise<DataQueryResult>;
}
export interface DataClient {
readonly auth: Readonly<{
getUser: () => Promise<{
data: { user: { id: string; email?: string | null } | null };
error: unknown;
}>;
}>;
from(table: string): DataQueryBuilder;
rpc(fn: string, args?: Readonly<Record<string, unknown>>): Promise<DataQueryResult>;
}
@@ -1,4 +1,5 @@
import { Pool, type PoolClient } from "pg";
import type { DataClient, DataQueryBuilder } from "./data-client";
type LocalIdentity = Readonly<{ id: string; email: string | null }> | null;
export type LocalDatabaseRole = "authenticated" | "service_role";
@@ -457,7 +458,7 @@ export function queryValue(type: string | undefined, value: unknown): unknown {
return value;
}
class LocalPostgresQueryBuilder implements PromiseLike<QueryResult> {
class LocalPostgresQueryBuilder implements DataQueryBuilder {
private selectedColumns: string[] | null = null;
private mutation: Mutation | null = null;
private readonly filters: Filter[] = [];
@@ -863,7 +864,7 @@ function castType(value: string): string {
return value;
}
export class LocalPostgresDataClient {
export class LocalPostgresDataClient implements DataClient {
readonly auth: Readonly<{
getUser: () => Promise<
Readonly<{
@@ -8,7 +8,7 @@
*/
import { createHash } from "node:crypto";
import { requestCandidateIntervals, type DatedCandidateRange } from "../core/candidate-window.ts";
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { resolveMissingBirthTimezoneOffset } from "../../birth-profile-timezone.ts";
import { declaredClockRange } from "../../declared-birth-window.ts";
import {
@@ -43,7 +43,7 @@ import {
type RectificationEntrySummary,
} from "./open-request.ts";
type AccountingClient = SupabaseClient;
type AccountingClient = DataClient;
export type V9BaselineSnapshot = Readonly<{
birth_date: string;
@@ -1,4 +1,4 @@
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { RectificationCaseServiceError } from "./case-service.ts";
import type { OpenRectificationCaseRequest } from "./open-request.ts";
@@ -17,7 +17,7 @@ const domainsBySessionTheme: Readonly<Record<string, string>> = {
/** Only an owned persisted consultation is a product-domain source. */
export async function resolveRectificationProductDomain(
client: SupabaseClient,
client: DataClient,
userId: string,
request: OpenRectificationCaseRequest,
): Promise<string> {
@@ -44,7 +44,7 @@ import { decisionStateFingerprint } from "../core/decision-fingerprint";
const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
/**
* Minimal structural RPC client. SupabaseClient satisfies this; tests can
* Minimal structural RPC client. DataClient satisfies this; tests can
* pass a fake with a single rpc() method.
*/
export type RectificationRpcClient = {
@@ -1,6 +1,6 @@
import "server-only";
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { resolveFeaturePricing } from "@/lib/feature-pricing";
import { isProductEnabled } from "@/lib/product-access";
import { getRectificationEntrySummary } from "@/lib/rectification-agentic/v9/case-service";
@@ -27,7 +27,7 @@ import {
/** C3: whether the offer card may be shown, decided by the server and never by the model. */
export async function decideRectificationOffer(input: {
accounting: SupabaseClient;
accounting: DataClient;
userId: string;
subjectRole: "self" | "other";
modelId: string;
@@ -42,7 +42,7 @@ export async function decideRectificationOffer(input: {
type SessionRow = { id: string; messages: unknown; session_type?: string | null };
async function readConsultationSession(client: SupabaseClient, userId: string, sessionId: string): Promise<SessionRow | null> {
async function readConsultationSession(client: DataClient, userId: string, sessionId: string): Promise<SessionRow | null> {
const { data, error } = await client
.from("chat_sessions")
.select("id,messages,session_type")
@@ -54,7 +54,7 @@ async function readConsultationSession(client: SupabaseClient, userId: string, s
return (row.session_type ?? "consultation") === "consultation" ? row : null;
}
async function writeMessages(client: SupabaseClient, userId: string, sessionId: string, messages: unknown[]): Promise<boolean> {
async function writeMessages(client: DataClient, userId: string, sessionId: string, messages: unknown[]): Promise<boolean> {
const { error } = await client
.from("chat_sessions")
.update({ messages })
@@ -71,7 +71,7 @@ async function writeMessages(client: SupabaseClient, userId: string, sessionId:
* Does not touch consultation_requests.
*/
export async function persistOpenedDirectOffer(input: {
client: SupabaseClient;
client: DataClient;
userId: string;
sessionId: string;
sentence: string;
@@ -96,7 +96,7 @@ export async function persistOpenedDirectOffer(input: {
/** C4: after a case opens from a consultation, bind the offer card to it. Best effort. */
export async function bindRectificationSource(input: {
client: SupabaseClient;
client: DataClient;
userId: string;
sourceSessionId: string;
caseId: string;
@@ -118,7 +118,7 @@ const SOURCE_SEARCH_LIMIT = 20;
* from history has none.
*/
export async function findRectificationSource(
client: SupabaseClient,
client: DataClient,
userId: string,
caseId: string,
): Promise<RectificationSource | null> {
@@ -139,8 +139,8 @@ export async function findRectificationSource(
* never from the browser.
*/
export async function returnToRectificationSource(input: {
client: SupabaseClient;
accounting: SupabaseClient;
client: DataClient;
accounting: DataClient;
userId: string;
caseId: string;
now?: () => Date;
@@ -1,11 +1,12 @@
import { createClient, type SupabaseClient } from "@supabase/supabase-js";
import { createClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import { createLocalPostgresDataClient } from "@/lib/db/local-postgres-client-core";
import {
getSupabaseUrl,
SupabaseConfigurationError,
} from "./config";
export function createAdminSupabaseClient() {
export function createAdminSupabaseClient(): DataClient {
if (process.env.AUTH_PROVIDER?.trim() === "self-hosted") {
const connectionString = process.env.SERVICE_DATABASE_URL?.trim();
if (!connectionString) {
@@ -18,7 +19,7 @@ export function createAdminSupabaseClient() {
connectionString,
null,
"service_role",
) as unknown as SupabaseClient;
) as DataClient;
}
const url = getSupabaseUrl();
const serviceRoleKey = process.env.SUPABASE_SERVICE_ROLE_KEY;
@@ -26,9 +27,10 @@ export function createAdminSupabaseClient() {
throw new SupabaseConfigurationError(["SUPABASE_SERVICE_ROLE_KEY"]);
}
// 回滚分支,运行时形状是超集
return createClient(url, serviceRoleKey, {
auth: { autoRefreshToken: false, persistSession: false },
});
}) as unknown as DataClient;
}
export function isAdminEmail(email: string | null | undefined) {
+5 -4
View File
@@ -1,8 +1,8 @@
import "server-only";
import { createServerClient } from "@supabase/ssr";
import type { SupabaseClient } from "@supabase/supabase-js";
import { cookies, headers } from "next/headers";
import type { DataClient } from "@/lib/db/data-client";
import { createLocalPostgresDataClient } from "@/lib/db/local-postgres-client";
import { readDatabaseUrl } from "@/lib/db/config";
import { getIdentityAuthServices } from "@/modules/identity/auth";
@@ -11,7 +11,7 @@ import { getSupabasePublicConfig } from "./config";
export async function createServerSupabaseClient(options?: {
readonly disableCookieCache?: boolean;
}) {
}): Promise<DataClient> {
if (process.env.AUTH_PROVIDER?.trim() === "self-hosted") {
const requestHeaders = new Headers(await headers());
const session = await readIdentitySession(
@@ -22,11 +22,12 @@ export async function createServerSupabaseClient(options?: {
return createLocalPostgresDataClient(
readDatabaseUrl(process.env, "APP_DATABASE_URL"),
session ? { id: session.user.id, email: session.user.email } : null,
) as unknown as SupabaseClient;
) as DataClient;
}
const { url, anonKey } = getSupabasePublicConfig();
const cookieStore = await cookies();
// 回滚分支,运行时形状是超集
return createServerClient(url, anonKey, {
cookies: {
getAll: () => cookieStore.getAll(),
@@ -40,5 +41,5 @@ export async function createServerSupabaseClient(options?: {
}
},
},
});
}) as unknown as DataClient;
}
@@ -12,7 +12,7 @@
import { windowIntervalsWidth } from "../lib/rectification-agentic/core/candidate-window.ts";
import { createTool } from "@mastra/core/tools";
import { z } from "zod";
import type { SupabaseClient } from "@supabase/supabase-js";
import type { DataClient } from "@/lib/db/data-client";
import type { PublicRectificationMethod } from "@/lib/rectification-agentic/v9/public-receipt";
import {
canonicalToolInputFingerprint,
@@ -196,7 +196,7 @@ export type RectificationV9Context = Readonly<{
turnId: string;
attemptId?: string;
userMessage?: string | null;
accounting: SupabaseClient;
accounting: DataClient;
engineBase?: string;
decision?: RectificationDecision;
productDelivery?: import("../lib/rectification-agentic/v9/product-delivery.ts").ProductDelivery | null;
@@ -0,0 +1,157 @@
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import test from "node:test";
import {
closeLocalPostgresDataPools,
createLocalPostgresDataClient,
} from "../src/lib/db/local-postgres-client-core.ts";
import type { DataClient } from "../src/lib/db/data-client.ts";
import { birthTimeScoringAlgorithmVersion } from "../src/lib/birth-time-scoring-job.ts";
import { createSupabaseScoringJobStore } from "../src/lib/birth-time-scoring-job-store.ts";
import { startPostgresFixture } from "./helpers/postgres-fixture.ts";
const runnerPath = fileURLToPath(new URL("../scripts/db-migrate.mjs", import.meta.url));
function dockerAvailable(): boolean {
return spawnSync("docker", ["version", "--format", "{{.Server.Version}}"], {
encoding: "utf8",
stdio: "ignore",
}).status === 0;
}
const skipWithoutDocker = dockerAvailable() ? false : "docker unavailable on this host";
const ids = {
user: "e1111111-1111-4111-8111-111111111111",
caseId: "e2222222-2222-4222-8222-222222222222",
job: "e3333333-3333-4333-8333-333333333333",
shapeJob: "e4444444-4444-4444-8444-444444444444",
};
const fingerprint = "claim-shape-fingerprint";
function sqlLiteral(value: string): string {
return `'${value.replaceAll("'", "''")}'`;
}
const identity = {
userId: ids.user,
caseId: ids.caseId,
jobId: ids.job,
evidenceFingerprint: fingerprint,
algorithmVersion: birthTimeScoringAlgorithmVersion,
now: "2026-10-15T00:00:00.000Z",
};
test("claimScoringJob reads the first row when rpc returns an array", async () => {
const rpc = async () => ({
data: [{ claim_state: "claimed", algorithm_version: birthTimeScoringAlgorithmVersion }],
error: null,
});
const store = createSupabaseScoringJobStore({ rpc } as unknown as DataClient, async () => null);
const claim = await store.claimScoringJob(identity);
assert.deepEqual(claim, {
kind: "claimed",
algorithmVersion: birthTimeScoringAlgorithmVersion,
});
});
test("claimScoringJob also accepts the single-object rpc shape", async () => {
const rpc = async () => ({
data: { claim_state: "processing", algorithm_version: birthTimeScoringAlgorithmVersion },
error: null,
});
const store = createSupabaseScoringJobStore({ rpc } as unknown as DataClient, async () => null);
const claim = await store.claimScoringJob(identity);
assert.deepEqual(claim, {
kind: "processing",
algorithmVersion: birthTimeScoringAlgorithmVersion,
});
});
test("claim_birth_time_scoring_job returns an array and claimScoringJob accepts it", { skip: skipWithoutDocker }, async () => {
const fixture = startPostgresFixture();
const schemaUrl = fixture.connectionUrl("schema_owner", "schema-owner-test-password");
try {
const migration = spawnSync(process.execPath, [runnerPath], {
encoding: "utf8",
env: { ...process.env, SCHEMA_DATABASE_URL: schemaUrl },
});
assert.equal(
migration.status,
0,
JSON.stringify({
status: migration.status,
stderr: migration.stderr,
stdoutTail: String(migration.stdout).slice(-1500),
}),
);
fixture.psqlAs(
"identity_runtime",
"identity-runtime-test-password",
`insert into identity.users (id, name, email, email_verified, email_verified_at)
values (${sqlLiteral(ids.user)}, 'Claim Shape', 'claim-shape@example.com', true, now());`,
);
fixture.psql(`
insert into public.birth_time_rectification_cases (
id, user_id, status, reported_date, source
) values (
${sqlLiteral(ids.caseId)}, ${sqlLiteral(ids.user)}, 'rectifying', '1990-01-01', 'family_exact'
);
insert into public.birth_time_rectification_scoring_jobs (
id, case_id, user_id, evidence_fingerprint, algorithm_version, status, expires_at
) values (
${sqlLiteral(ids.job)}, ${sqlLiteral(ids.caseId)}, ${sqlLiteral(ids.user)},
${sqlLiteral(fingerprint)}, ${sqlLiteral(birthTimeScoringAlgorithmVersion)},
'pending', '2026-10-15T01:00:00Z'
), (
${sqlLiteral(ids.shapeJob)}, ${sqlLiteral(ids.caseId)}, ${sqlLiteral(ids.user)},
${sqlLiteral(`${fingerprint}-shape`)}, ${sqlLiteral(birthTimeScoringAlgorithmVersion)},
'pending', '2026-10-15T01:00:00Z'
);
update public.birth_time_rectification_cases
set journey_protocol = 'legacy-guided-v1',
turn_state = ${sqlLiteral(JSON.stringify({
nextAction: { kind: "score_pending", jobId: ids.shapeJob },
}))}::jsonb
where id = ${sqlLiteral(ids.caseId)};
`);
const client = createLocalPostgresDataClient(
fixture.connectionUrl("service_runtime", "service-runtime-test-password"),
null,
"service_role",
);
const direct = await client.rpc("claim_birth_time_scoring_job", {
p_user_id: ids.user,
p_case_id: ids.caseId,
p_job_id: ids.shapeJob,
p_evidence_fingerprint: `${fingerprint}-shape`,
p_algorithm_version: birthTimeScoringAlgorithmVersion,
p_now: identity.now,
});
assert.equal(direct.error, null, direct.error?.message);
assert.equal(Array.isArray(direct.data), true);
assert.equal(direct.data[0]?.claim_state, "claimed");
fixture.psql(`
update public.birth_time_rectification_cases
set turn_state = ${sqlLiteral(JSON.stringify({
nextAction: { kind: "score_pending", jobId: ids.job },
}))}::jsonb
where id = ${sqlLiteral(ids.caseId)};
`);
const store = createSupabaseScoringJobStore(client, async () => null);
const claim = await store.claimScoringJob(identity);
assert.deepEqual(claim, {
kind: "claimed",
algorithmVersion: birthTimeScoringAlgorithmVersion,
});
} finally {
await closeLocalPostgresDataPools();
fixture.stop();
}
});