jesse-ux
e4e73f56c0
对话、星盘、星历、报告进同一 (app) 外壳，列表只拉一次。服务端不再列出空咨询；新建复用已有空会话。新标题改成「生时校正 · M月D日」，侧栏副标题用创建时间。
747 lines
23 KiB
TypeScript
747 lines
23 KiB
TypeScript
import { Pool, type PoolClient } from "pg";
|
|
|
|
type LocalIdentity = Readonly<{ id: string; email: string | null }> | null;
|
|
export type LocalDatabaseRole = "authenticated" | "service_role";
|
|
type PostgresError = Error & { code?: string };
|
|
type QueryError = Readonly<{ message: string; code?: string }>;
|
|
type QueryResult = Readonly<{
|
|
data: unknown;
|
|
error: QueryError | null;
|
|
count?: number | null;
|
|
}>;
|
|
|
|
type CmpOp = "eq" | "neq" | "lt" | "gt" | "lte" | "gte";
|
|
|
|
type Filter =
|
|
| Readonly<{ kind: "eq"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "neq"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "gt"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "lte"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "like"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "in"; column: string; value: readonly unknown[] }>
|
|
| Readonly<{ kind: "is"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "notContains"; column: string; value: unknown }>
|
|
| Readonly<{ kind: "or"; expression: string }>;
|
|
|
|
export type PostgrestOrNode =
|
|
| Readonly<{ kind: "cmp"; column: string; op: CmpOp; value: unknown }>
|
|
| Readonly<{ kind: "and"; nodes: readonly PostgrestOrNode[] }>;
|
|
|
|
type Mutation =
|
|
| Readonly<{ kind: "insert"; rows: readonly Record<string, unknown>[] }>
|
|
| Readonly<{ kind: "update"; values: Record<string, unknown> }>
|
|
| Readonly<{
|
|
kind: "upsert";
|
|
rows: readonly Record<string, unknown>[];
|
|
conflict: readonly string[];
|
|
}>
|
|
| Readonly<{ kind: "delete"; exactCount: boolean }>;
|
|
|
|
const identifierPattern = /^[a-z_][a-z0-9_]*$/;
|
|
|
|
function identifier(value: string): string {
|
|
const normalized = value.trim();
|
|
if (!identifierPattern.test(normalized))
|
|
throw new Error("unsafe database identifier");
|
|
return `"${normalized}"`;
|
|
}
|
|
|
|
export function formatOrderClause(
|
|
ordering: readonly { column: string; ascending: boolean }[],
|
|
): string {
|
|
if (ordering.length === 0) return "";
|
|
return ` order by ${ordering
|
|
.map((item) => `${identifier(item.column)} ${item.ascending ? "asc" : "desc"}`)
|
|
.join(", ")}`;
|
|
}
|
|
|
|
const SQL_CMP: Record<CmpOp, string> = {
|
|
eq: "=",
|
|
neq: "<>",
|
|
lt: "<",
|
|
gt: ">",
|
|
lte: "<=",
|
|
gte: ">=",
|
|
};
|
|
|
|
function splitTopLevel(expression: string): string[] {
|
|
const parts: string[] = [];
|
|
let depth = 0;
|
|
let inString = false;
|
|
let start = 0;
|
|
for (let index = 0; index < expression.length; index += 1) {
|
|
const char = expression[index];
|
|
if (char === "\"" && expression[index - 1] !== "\\") inString = !inString;
|
|
if (inString) continue;
|
|
if (char === "(") depth += 1;
|
|
else if (char === ")") depth -= 1;
|
|
else if (char === "," && depth === 0) {
|
|
parts.push(expression.slice(start, index));
|
|
start = index + 1;
|
|
}
|
|
}
|
|
parts.push(expression.slice(start));
|
|
return parts.map((part) => part.trim()).filter(Boolean);
|
|
}
|
|
|
|
function parsePostgrestValue(raw: string): unknown {
|
|
const trimmed = raw.trim();
|
|
if (trimmed === "true") return true;
|
|
if (trimmed === "false") return false;
|
|
if (trimmed === "null") return null;
|
|
if (trimmed.startsWith("\"") || trimmed.startsWith("[") || trimmed.startsWith("{")) {
|
|
return JSON.parse(trimmed) as unknown;
|
|
}
|
|
return trimmed;
|
|
}
|
|
|
|
function parsePostgrestTerm(term: string): PostgrestOrNode {
|
|
const andMatch = /^and\(([\s\S]*)\)$/.exec(term);
|
|
if (andMatch?.[1] != null) {
|
|
return { kind: "and", nodes: splitTopLevel(andMatch[1]).map(parsePostgrestTerm) };
|
|
}
|
|
const cmp = /^([a-z_][a-z0-9_]*)\.(eq|neq|lt|gt|lte|gte)\.([\s\S]*)$/.exec(term);
|
|
if (!cmp) throw new Error(`unsupported or filter: ${term}`);
|
|
return {
|
|
kind: "cmp",
|
|
column: cmp[1] ?? "",
|
|
op: (cmp[2] ?? "eq") as CmpOp,
|
|
value: parsePostgrestValue(cmp[3] ?? ""),
|
|
};
|
|
}
|
|
|
|
export function parsePostgrestOr(expression: string): PostgrestOrNode[] {
|
|
return splitTopLevel(expression).map(parsePostgrestTerm);
|
|
}
|
|
|
|
function compilePostgrestNode(
|
|
node: PostgrestOrNode,
|
|
parameters: unknown[],
|
|
types: Map<string, string>,
|
|
): string {
|
|
if (node.kind === "and") {
|
|
return `(${node.nodes.map((child) => compilePostgrestNode(child, parameters, types)).join(" and ")})`;
|
|
}
|
|
identifier(node.column);
|
|
parameters.push(databaseValue(types.get(node.column), node.value));
|
|
return `${identifier(node.column)} ${SQL_CMP[node.op]} $${parameters.length}`;
|
|
}
|
|
|
|
export function compilePostgrestOr(
|
|
expression: string,
|
|
parameters: unknown[],
|
|
types: Map<string, string>,
|
|
): string {
|
|
const nodes = parsePostgrestOr(expression);
|
|
if (nodes.length === 0) return "true";
|
|
if (nodes.length === 1) {
|
|
const only = nodes[0];
|
|
if (!only) return "true";
|
|
return compilePostgrestNode(only, parameters, types);
|
|
}
|
|
return `(${nodes.map((node) => compilePostgrestNode(node, parameters, types)).join(" or ")})`;
|
|
}
|
|
|
|
export function upsertConflictColumns(options?: { onConflict?: string }): string[] {
|
|
return (options?.onConflict ?? "")
|
|
.split(",")
|
|
.map((column) => column.trim())
|
|
.filter(Boolean);
|
|
}
|
|
|
|
async function primaryKeyColumns(client: PoolClient, table: string): Promise<string[]> {
|
|
identifier(table);
|
|
const result = await client.query<{ column_name: string }>(
|
|
`
|
|
select a.attname as column_name
|
|
from pg_index i
|
|
join pg_attribute a on a.attrelid = i.indrelid and a.attnum = any(i.indkey)
|
|
where i.indrelid = format('%I.%I', 'public', $1::text)::regclass
|
|
and i.indisprimary
|
|
order by array_position(i.indkey, a.attnum)
|
|
`,
|
|
[table],
|
|
);
|
|
return result.rows.map((row) => row.column_name);
|
|
}
|
|
|
|
function queryError(error: unknown): QueryError {
|
|
const value = error as PostgresError;
|
|
return {
|
|
message: value instanceof Error ? value.message : "database request failed",
|
|
...(typeof value?.code === "string" ? { code: value.code } : {}),
|
|
};
|
|
}
|
|
|
|
function records(
|
|
value: Record<string, unknown> | readonly Record<string, unknown>[],
|
|
) {
|
|
return Array.isArray(value) ? value : [value];
|
|
}
|
|
|
|
const poolGlobal = globalThis as typeof globalThis & {
|
|
jyotishaLocalDataPools?: Map<string, Pool>;
|
|
};
|
|
|
|
function localDataPool(connectionString: string): Pool {
|
|
poolGlobal.jyotishaLocalDataPools ??= new Map();
|
|
let pool = poolGlobal.jyotishaLocalDataPools.get(connectionString);
|
|
if (!pool) {
|
|
pool = new Pool({
|
|
connectionString,
|
|
max: 10,
|
|
idleTimeoutMillis: 30_000,
|
|
connectionTimeoutMillis: 5_000,
|
|
allowExitOnIdle: true,
|
|
application_name: "jyotisha-business",
|
|
});
|
|
poolGlobal.jyotishaLocalDataPools.set(connectionString, pool);
|
|
}
|
|
return pool;
|
|
}
|
|
|
|
export async function closeLocalPostgresDataPool(connectionString: string): Promise<void> {
|
|
const pools = poolGlobal.jyotishaLocalDataPools;
|
|
const pool = pools?.get(connectionString);
|
|
if (!pool) return;
|
|
pools?.delete(connectionString);
|
|
await pool.end();
|
|
}
|
|
|
|
export async function closeLocalPostgresDataPools(): Promise<void> {
|
|
const pools = poolGlobal.jyotishaLocalDataPools;
|
|
poolGlobal.jyotishaLocalDataPools = new Map();
|
|
if (!pools) return;
|
|
await Promise.all([...pools.values()].map((pool) => pool.end()));
|
|
}
|
|
|
|
async function inBusinessTransaction<T>(
|
|
pool: Pool,
|
|
identity: LocalIdentity,
|
|
role: LocalDatabaseRole | null,
|
|
run: (client: PoolClient) => Promise<T>,
|
|
): Promise<T> {
|
|
const client = await pool.connect();
|
|
try {
|
|
await client.query("begin");
|
|
if (role !== null) await client.query(`set local role ${role}`);
|
|
await client.query(
|
|
"select set_config('request.jwt.claim.sub', $1, true), set_config('request.jwt.claim.email', $2, true)",
|
|
[identity?.id ?? "", identity?.email ?? ""],
|
|
);
|
|
const result = await run(client);
|
|
await client.query("commit");
|
|
return result;
|
|
} catch (error) {
|
|
await client.query("rollback").catch(() => undefined);
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|
|
|
|
async function columnTypes(
|
|
client: PoolClient,
|
|
table: string,
|
|
): Promise<Map<string, string>> {
|
|
const result = await client.query<{ column_name: string; udt_name: string }>(
|
|
`
|
|
select column_name, udt_name
|
|
from information_schema.columns
|
|
where table_schema = 'public' and table_name = $1
|
|
`,
|
|
[table],
|
|
);
|
|
return new Map(result.rows.map((row) => [row.column_name, row.udt_name]));
|
|
}
|
|
|
|
function databaseValue(type: string | undefined, value: unknown): unknown {
|
|
if ((type === "json" || type === "jsonb") && value !== null) {
|
|
return JSON.stringify(value);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
export function queryValue(type: string | undefined, value: unknown): unknown {
|
|
if (!(value instanceof Date) || !Number.isFinite(value.getTime())) return value;
|
|
if (type === "date") {
|
|
// pg parses DATE at local midnight; UTC formatting can shift the calendar day.
|
|
return [
|
|
String(value.getFullYear()).padStart(4, "0"),
|
|
String(value.getMonth() + 1).padStart(2, "0"),
|
|
String(value.getDate()).padStart(2, "0"),
|
|
].join("-");
|
|
}
|
|
if (
|
|
type === "timestamptz"
|
|
|| type === "timestamp"
|
|
|| type === "timestamp with time zone"
|
|
|| type === "timestamp without time zone"
|
|
) {
|
|
return value.toISOString();
|
|
}
|
|
return value;
|
|
}
|
|
|
|
class LocalPostgresQueryBuilder implements PromiseLike<QueryResult> {
|
|
private selectedColumns: string[] | null = null;
|
|
private mutation: Mutation | null = null;
|
|
private readonly filters: Filter[] = [];
|
|
private ordering: Array<{ column: string; ascending: boolean }> = [];
|
|
private rowLimit: number | null = null;
|
|
private abort: AbortSignal | null = null;
|
|
private cardinality: "many" | "single" | "maybeSingle" = "many";
|
|
|
|
constructor(
|
|
private readonly pool: Pool,
|
|
private readonly identity: LocalIdentity,
|
|
private readonly role: LocalDatabaseRole | null,
|
|
private readonly table: string,
|
|
) {
|
|
identifier(table);
|
|
}
|
|
|
|
select(columns = "*") {
|
|
this.selectedColumns =
|
|
columns === "*"
|
|
? ["*"]
|
|
: columns
|
|
.split(",")
|
|
.map((column) => column.trim())
|
|
.filter(Boolean);
|
|
for (const column of this.selectedColumns) {
|
|
if (column !== "*") identifier(column);
|
|
}
|
|
return this;
|
|
}
|
|
|
|
insert(value: Record<string, unknown> | readonly Record<string, unknown>[]) {
|
|
this.mutation = { kind: "insert", rows: records(value) };
|
|
return this;
|
|
}
|
|
|
|
upsert(
|
|
value: Record<string, unknown> | readonly Record<string, unknown>[],
|
|
options?: { onConflict?: string },
|
|
) {
|
|
const conflict = upsertConflictColumns(options);
|
|
conflict.forEach(identifier);
|
|
this.mutation = { kind: "upsert", rows: records(value), conflict };
|
|
return this;
|
|
}
|
|
|
|
update(values: Record<string, unknown>) {
|
|
this.mutation = { kind: "update", values };
|
|
return this;
|
|
}
|
|
|
|
delete(options?: { count?: string }) {
|
|
this.mutation = { kind: "delete", exactCount: options?.count === "exact" };
|
|
return this;
|
|
}
|
|
|
|
eq(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "eq", column, value });
|
|
return this;
|
|
}
|
|
|
|
neq(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "neq", column, value });
|
|
return this;
|
|
}
|
|
|
|
gt(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "gt", column, value });
|
|
return this;
|
|
}
|
|
|
|
lte(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "lte", column, value });
|
|
return this;
|
|
}
|
|
|
|
like(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "like", column, value });
|
|
return this;
|
|
}
|
|
|
|
in(column: string, value: readonly unknown[]) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "in", column, value });
|
|
return this;
|
|
}
|
|
|
|
is(column: string, value: unknown) {
|
|
identifier(column);
|
|
this.filters.push({ kind: "is", column, value });
|
|
return this;
|
|
}
|
|
|
|
not(column: string, operator: string, value: unknown) {
|
|
identifier(column);
|
|
if (operator === "eq") {
|
|
this.filters.push({ kind: "neq", column, value });
|
|
return this;
|
|
}
|
|
if (operator !== "cs") throw new Error("unsupported not filter");
|
|
this.filters.push({ kind: "notContains", column, value });
|
|
return this;
|
|
}
|
|
|
|
or(expression: string) {
|
|
if (!expression.trim()) throw new Error("empty or filter");
|
|
this.filters.push({ kind: "or", expression });
|
|
return this;
|
|
}
|
|
|
|
order(column: string, options: { ascending?: boolean } = {}) {
|
|
identifier(column);
|
|
this.ordering.push({ column, ascending: options.ascending !== false });
|
|
return this;
|
|
}
|
|
|
|
limit(value: number) {
|
|
if (!Number.isSafeInteger(value) || value < 0)
|
|
throw new Error("invalid row limit");
|
|
this.rowLimit = value;
|
|
return this;
|
|
}
|
|
|
|
abortSignal(signal: AbortSignal) {
|
|
this.abort = signal;
|
|
return this;
|
|
}
|
|
|
|
single() {
|
|
this.cardinality = "single";
|
|
return this.execute();
|
|
}
|
|
|
|
maybeSingle() {
|
|
this.cardinality = "maybeSingle";
|
|
return this.execute();
|
|
}
|
|
|
|
then<TResult1 = QueryResult, TResult2 = never>(
|
|
onfulfilled?:
|
|
((value: QueryResult) => TResult1 | PromiseLike<TResult1>) | null,
|
|
onrejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
|
|
): PromiseLike<TResult1 | TResult2> {
|
|
return this.execute().then(onfulfilled, onrejected);
|
|
}
|
|
|
|
private returningClause(): string {
|
|
if (!this.selectedColumns) return "";
|
|
return ` returning ${this.selectedColumns.map((column) => (column === "*" ? "*" : identifier(column))).join(", ")}`;
|
|
}
|
|
|
|
private filterClause(
|
|
parameters: unknown[],
|
|
types: Map<string, string>,
|
|
): string {
|
|
if (this.filters.length === 0) return "";
|
|
const parts = this.filters.map((filter) => {
|
|
if (filter.kind === "or") {
|
|
return compilePostgrestOr(filter.expression, parameters, types);
|
|
}
|
|
const column = identifier(filter.column);
|
|
if (filter.kind === "is") {
|
|
if (filter.value === null) return `${column} is null`;
|
|
if (filter.value === true) return `${column} is true`;
|
|
if (filter.value === false) return `${column} is false`;
|
|
throw new Error("unsupported is filter");
|
|
}
|
|
if (filter.kind === "in") {
|
|
if (filter.value.length === 0) return "false";
|
|
const placeholders = filter.value.map((value) => {
|
|
parameters.push(databaseValue(types.get(filter.column), value));
|
|
return `$${parameters.length}`;
|
|
});
|
|
return `${column} in (${placeholders.join(", ")})`;
|
|
}
|
|
if (filter.kind === "notContains") {
|
|
parameters.push(databaseValue(types.get(filter.column), filter.value));
|
|
return `not (${column} @> $${parameters.length})`;
|
|
}
|
|
parameters.push(databaseValue(types.get(filter.column), filter.value));
|
|
const operator =
|
|
filter.kind === "neq"
|
|
? "<>"
|
|
: filter.kind === "gt"
|
|
? ">"
|
|
: filter.kind === "lte"
|
|
? "<="
|
|
: filter.kind === "like"
|
|
? "like"
|
|
: "=";
|
|
return `${column} ${operator} $${parameters.length}`;
|
|
});
|
|
return ` where ${parts.join(" and ")}`;
|
|
}
|
|
|
|
private async execute(): Promise<QueryResult> {
|
|
if (this.abort?.aborted) {
|
|
return { data: null, error: { message: "AbortError" } };
|
|
}
|
|
try {
|
|
return await inBusinessTransaction(
|
|
this.pool,
|
|
this.identity,
|
|
this.role,
|
|
async (client) => {
|
|
const types = await columnTypes(client, this.table);
|
|
const parameters: unknown[] = [];
|
|
let sql: string;
|
|
|
|
if (!this.mutation) {
|
|
const selected = (this.selectedColumns ?? ["*"])
|
|
.map((column) => (column === "*" ? "*" : identifier(column)))
|
|
.join(", ");
|
|
sql = `select ${selected} from public.${identifier(this.table)}`;
|
|
sql += this.filterClause(parameters, types);
|
|
sql += formatOrderClause(this.ordering);
|
|
if (this.rowLimit !== null) sql += ` limit ${this.rowLimit}`;
|
|
} else if (
|
|
this.mutation.kind === "insert" ||
|
|
this.mutation.kind === "upsert"
|
|
) {
|
|
const rows = this.mutation.rows;
|
|
if (rows.length === 0)
|
|
return { data: this.selectedColumns ? [] : null, error: null };
|
|
const columns = Object.keys(rows[0] ?? {});
|
|
if (
|
|
columns.length === 0 ||
|
|
rows.some(
|
|
(row) => Object.keys(row).join("\0") !== columns.join("\0"),
|
|
)
|
|
) {
|
|
throw new Error("inconsistent insert rows");
|
|
}
|
|
columns.forEach(identifier);
|
|
const valueGroups = rows.map(
|
|
(row) =>
|
|
`(${columns
|
|
.map((column) => {
|
|
parameters.push(
|
|
databaseValue(types.get(column), row[column]),
|
|
);
|
|
return `$${parameters.length}`;
|
|
})
|
|
.join(", ")})`,
|
|
);
|
|
sql = `insert into public.${identifier(this.table)} (${columns.map(identifier).join(", ")}) values ${valueGroups.join(", ")}`;
|
|
if (this.mutation.kind === "upsert") {
|
|
let conflict = this.mutation.conflict;
|
|
if (conflict.length === 0) {
|
|
conflict = await primaryKeyColumns(client, this.table);
|
|
}
|
|
if (conflict.length === 0) {
|
|
throw new Error("upsert requires a conflict target");
|
|
}
|
|
const updates = columns.filter((column) => !conflict.includes(column));
|
|
sql += ` on conflict (${conflict.map(identifier).join(", ")}) do ${
|
|
updates.length === 0
|
|
? "nothing"
|
|
: `update set ${updates.map((column) => `${identifier(column)} = excluded.${identifier(column)}`).join(", ")}`
|
|
}`;
|
|
}
|
|
sql += this.returningClause();
|
|
} else if (this.mutation.kind === "update") {
|
|
const columns = Object.keys(this.mutation.values);
|
|
if (columns.length === 0) throw new Error("empty update");
|
|
const assignments = columns.map((column) => {
|
|
identifier(column);
|
|
parameters.push(
|
|
databaseValue(
|
|
types.get(column),
|
|
this.mutation && this.mutation.kind === "update"
|
|
? this.mutation.values[column]
|
|
: null,
|
|
),
|
|
);
|
|
return `${identifier(column)} = $${parameters.length}`;
|
|
});
|
|
sql = `update public.${identifier(this.table)} set ${assignments.join(", ")}`;
|
|
sql += this.filterClause(parameters, types);
|
|
sql += this.returningClause();
|
|
} else {
|
|
sql = `delete from public.${identifier(this.table)}`;
|
|
sql += this.filterClause(parameters, types);
|
|
sql += this.returningClause();
|
|
}
|
|
|
|
const result = await client.query(sql, parameters);
|
|
const rows = result.rows.map((row) =>
|
|
Object.fromEntries(
|
|
Object.entries(row).map(([column, value]) => [
|
|
column,
|
|
queryValue(types.get(column), value),
|
|
]),
|
|
),
|
|
);
|
|
let data: unknown = this.selectedColumns ? rows : null;
|
|
if (this.cardinality !== "many") {
|
|
if (
|
|
rows.length > 1 ||
|
|
(this.cardinality === "single" && rows.length !== 1)
|
|
) {
|
|
return {
|
|
data: null,
|
|
error: { code: "PGRST116", message: "unexpected row count" },
|
|
};
|
|
}
|
|
data = rows[0] ?? null;
|
|
}
|
|
return {
|
|
data,
|
|
error: null,
|
|
...(this.mutation?.kind === "delete" && this.mutation.exactCount
|
|
? { count: result.rowCount ?? 0 }
|
|
: {}),
|
|
};
|
|
},
|
|
);
|
|
} catch (error) {
|
|
return { data: null, error: queryError(error), count: null };
|
|
}
|
|
}
|
|
}
|
|
|
|
type FunctionMetadata = Readonly<{
|
|
proretset: boolean;
|
|
return_type: string;
|
|
argument_names: string[] | null;
|
|
argument_types: string[];
|
|
}>;
|
|
|
|
async function functionMetadata(
|
|
client: PoolClient,
|
|
functionName: string,
|
|
argumentNames: readonly string[],
|
|
): Promise<FunctionMetadata> {
|
|
const result = await client.query<FunctionMetadata>(
|
|
`
|
|
select
|
|
p.proretset,
|
|
format_type(p.prorettype, null) as return_type,
|
|
p.proargnames as argument_names,
|
|
array(
|
|
select format_type(argument_type, null)
|
|
from unnest(p.proargtypes) argument_type
|
|
) as argument_types
|
|
from pg_proc p
|
|
join pg_namespace n on n.oid = p.pronamespace
|
|
where n.nspname = 'public'
|
|
and p.proname = $1
|
|
and $2::text[] <@ coalesce(p.proargnames, '{}'::text[])
|
|
order by cardinality(p.proargtypes) asc
|
|
limit 1
|
|
`,
|
|
[functionName, argumentNames],
|
|
);
|
|
const metadata = result.rows[0];
|
|
if (!metadata) throw new Error("database function not found");
|
|
return metadata;
|
|
}
|
|
|
|
function castType(value: string): string {
|
|
if (!/^[a-z0-9_ .\[\]]+$/.test(value))
|
|
throw new Error("unsafe database type");
|
|
return value;
|
|
}
|
|
|
|
export class LocalPostgresDataClient {
|
|
readonly auth: Readonly<{
|
|
getUser: () => Promise<
|
|
Readonly<{
|
|
data: { user: LocalIdentity };
|
|
error: null;
|
|
}>
|
|
>;
|
|
}>;
|
|
|
|
private readonly pool: Pool;
|
|
|
|
constructor(
|
|
connectionString: string,
|
|
private readonly identity: LocalIdentity,
|
|
private readonly role: LocalDatabaseRole | null,
|
|
) {
|
|
if (role !== null && role !== "authenticated" && role !== "service_role") {
|
|
throw new Error("unsupported database role");
|
|
}
|
|
this.pool = localDataPool(connectionString);
|
|
this.auth = {
|
|
getUser: async () => ({ data: { user: this.identity }, error: null }),
|
|
};
|
|
}
|
|
|
|
from(table: string) {
|
|
return new LocalPostgresQueryBuilder(
|
|
this.pool,
|
|
this.identity,
|
|
this.role,
|
|
table,
|
|
);
|
|
}
|
|
|
|
async rpc(
|
|
functionName: string,
|
|
args: Readonly<Record<string, unknown>> = {},
|
|
) {
|
|
try {
|
|
identifier(functionName);
|
|
return await inBusinessTransaction(
|
|
this.pool,
|
|
this.identity,
|
|
this.role,
|
|
async (client) => {
|
|
const names = Object.keys(args);
|
|
names.forEach(identifier);
|
|
const metadata = await functionMetadata(client, functionName, names);
|
|
const typeByName = new Map(
|
|
(metadata.argument_names ?? []).map((name, index) => [
|
|
name,
|
|
metadata.argument_types[index],
|
|
]),
|
|
);
|
|
const parameters = names.map((name) =>
|
|
databaseValue(typeByName.get(name), args[name]),
|
|
);
|
|
const call = names
|
|
.map(
|
|
(name, index) =>
|
|
`${identifier(name)} => $${index + 1}::${castType(typeByName.get(name) ?? "text")}`,
|
|
)
|
|
.join(", ");
|
|
const sql = metadata.proretset
|
|
? `select * from public.${identifier(functionName)}(${call})`
|
|
: `select public.${identifier(functionName)}(${call}) as value`;
|
|
const result = await client.query(sql, parameters);
|
|
return {
|
|
data: metadata.proretset
|
|
? result.rows
|
|
: (result.rows[0]?.value ?? null),
|
|
error: null,
|
|
};
|
|
},
|
|
);
|
|
} catch (error) {
|
|
return { data: null, error: queryError(error) };
|
|
}
|
|
}
|
|
}
|
|
|
|
export function createLocalPostgresDataClient(
|
|
connectionString: string,
|
|
identity: LocalIdentity = null,
|
|
role: LocalDatabaseRole | null = "authenticated",
|
|
) {
|
|
return new LocalPostgresDataClient(connectionString, identity, role);
|
|
}
|