303 lines
11 KiB
TypeScript
303 lines
11 KiB
TypeScript
import type { Db, ExecResult, Row } from "./driver.ts";
|
|
import type { Model } from "./schema.ts";
|
|
import type { Dialect } from "./sql.ts";
|
|
|
|
const IDENTIFIER = /^[A-Za-z_][A-Za-z0-9_]*$/;
|
|
|
|
function identifier(value: string): string {
|
|
if (!IDENTIFIER.test(value)) throw new TypeError(`Unsafe SQL identifier: ${value}`);
|
|
return value;
|
|
}
|
|
|
|
function placeholder(dialect: Dialect, index: number): string {
|
|
return dialect === "postgres" ? `$${index}` : "?";
|
|
}
|
|
|
|
function placeholders(dialect: Dialect, count: number, start = 1): string[] {
|
|
return Array.from({ length: count }, (_value, index) => placeholder(dialect, start + index));
|
|
}
|
|
|
|
function finiteInteger(value: number | undefined, fallback: number, minimum: number): number {
|
|
if (value === undefined) return fallback;
|
|
if (!Number.isFinite(value)) throw new RangeError("Expected a finite integer.");
|
|
return Math.max(minimum, Math.floor(value));
|
|
}
|
|
|
|
export class RecordNotFoundError extends Error {
|
|
constructor(message = "Record not found") {
|
|
super(message);
|
|
this.name = "RecordNotFoundError";
|
|
}
|
|
}
|
|
|
|
export async function firstOrThrow<T>(
|
|
db: Db,
|
|
sql: string,
|
|
params: unknown[] = [],
|
|
model?: Model<T>,
|
|
message?: string,
|
|
): Promise<T> {
|
|
const row = await db.one(sql, params, model);
|
|
if (row === null) throw new RecordNotFoundError(message);
|
|
return row;
|
|
}
|
|
|
|
export async function exists(db: Db, sql: string, params: unknown[] = []): Promise<boolean> {
|
|
return (await db.one(sql, params)) !== null;
|
|
}
|
|
|
|
export async function countRows(
|
|
db: Db,
|
|
table: string,
|
|
where = "",
|
|
params: unknown[] = [],
|
|
): Promise<number> {
|
|
const row = await db.one<{ count: number | string }>(
|
|
`SELECT COUNT(*) AS count FROM ${identifier(table)}${where ? ` WHERE ${where}` : ""}`,
|
|
params,
|
|
);
|
|
return Number(row?.count ?? 0);
|
|
}
|
|
|
|
export function withTransaction<T>(db: Db, callback: (tx: Db) => Promise<T>): Promise<T> {
|
|
return db.tx(callback);
|
|
}
|
|
|
|
/** Execute a parameterized compare-and-set update and report whether this caller won. */
|
|
export async function compareAndSet(
|
|
db: Db,
|
|
options: {
|
|
table: string;
|
|
idColumn?: string;
|
|
id: string | number;
|
|
stateColumn?: string;
|
|
from: string;
|
|
to: string;
|
|
extra?: Record<string, unknown>;
|
|
},
|
|
): Promise<boolean> {
|
|
const table = identifier(options.table);
|
|
const idColumn = identifier(options.idColumn ?? "id");
|
|
const stateColumn = identifier(options.stateColumn ?? "status");
|
|
const entries = Object.entries(options.extra ?? {});
|
|
const assignments = [stateColumn, ...entries.map(([name]) => identifier(name))];
|
|
const dialect = db.driver.dialect;
|
|
const values = [options.to, ...entries.map(([, value]) => value), options.id, options.from];
|
|
const sql = `UPDATE ${table} SET ${assignments
|
|
.map((name, index) => `${name} = ${placeholder(dialect, index + 1)}`)
|
|
.join(
|
|
", ",
|
|
)} WHERE ${idColumn} = ${placeholder(dialect, assignments.length + 1)} AND ${stateColumn} = ${placeholder(dialect, assignments.length + 2)}`;
|
|
return (await db.exec(sql, values)).changes === 1;
|
|
}
|
|
|
|
/** Conservative default classifier for deadlock/serialization retry errors. */
|
|
export function isRetryableTransactionError(error: unknown): boolean {
|
|
if (!error || typeof error !== "object") return false;
|
|
const value = error as { code?: unknown; errno?: unknown; message?: unknown };
|
|
const code = String(value.code ?? value.errno ?? "").toUpperCase();
|
|
if (["40001", "40P01", "SQLITE_BUSY", "SQLITE_LOCKED", "1213", "1205"].includes(code)) {
|
|
return true;
|
|
}
|
|
const message = String(value.message ?? "").toLowerCase();
|
|
return /deadlock|serialization failure|database is locked|lock wait timeout/.test(message);
|
|
}
|
|
|
|
export async function retryTransaction<T>(
|
|
db: Db,
|
|
callback: (tx: Db, attempt: number) => Promise<T>,
|
|
options: {
|
|
attempts?: number;
|
|
baseDelayMs?: number;
|
|
maxDelayMs?: number;
|
|
jitter?: boolean;
|
|
shouldRetry?: (error: unknown) => boolean;
|
|
} = {},
|
|
): Promise<T> {
|
|
const attempts = finiteInteger(options.attempts, 3, 1);
|
|
const baseDelayMs = finiteInteger(options.baseDelayMs, 25, 0);
|
|
const maxDelayMs = finiteInteger(options.maxDelayMs, 1_000, 0);
|
|
const shouldRetry = options.shouldRetry ?? isRetryableTransactionError;
|
|
let lastError: unknown;
|
|
for (let attempt = 1; attempt <= attempts; attempt++) {
|
|
try {
|
|
return await db.tx((tx) => callback(tx, attempt));
|
|
} catch (error) {
|
|
lastError = error;
|
|
if (attempt >= attempts || !shouldRetry(error)) throw error;
|
|
const exponential = Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1));
|
|
const delay =
|
|
options.jitter === false
|
|
? exponential
|
|
: Math.round(exponential * (0.5 + Math.random() * 0.5));
|
|
await new Promise((resolve) => setTimeout(resolve, delay));
|
|
}
|
|
}
|
|
throw lastError;
|
|
}
|
|
|
|
export function batch<T>(values: readonly T[], size = 100): T[][] {
|
|
const chunkSize = finiteInteger(size, 100, 1);
|
|
const chunks: T[][] = [];
|
|
for (let index = 0; index < values.length; index += chunkSize) {
|
|
chunks.push(values.slice(index, index + chunkSize));
|
|
}
|
|
return chunks;
|
|
}
|
|
|
|
export async function databaseHealth(
|
|
db: Db,
|
|
): Promise<{ ok: boolean; latencyMs: number; error?: string }> {
|
|
const start = performance.now();
|
|
try {
|
|
await db.one("SELECT 1 AS healthy");
|
|
return { ok: true, latencyMs: Math.round((performance.now() - start) * 100) / 100 };
|
|
} catch (error) {
|
|
return {
|
|
ok: false,
|
|
latencyMs: Math.round((performance.now() - start) * 100) / 100,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
};
|
|
}
|
|
}
|
|
|
|
export interface RepositoryListOptions<T extends Row> {
|
|
limit?: number;
|
|
offset?: number;
|
|
orderBy?: keyof T & string;
|
|
direction?: "asc" | "desc";
|
|
}
|
|
|
|
export interface Repository<T extends Row> {
|
|
all(options?: RepositoryListOptions<T>): Promise<T[]>;
|
|
find(id: string | number): Promise<T | null>;
|
|
require(id: string | number): Promise<T>;
|
|
create(values: Partial<T>): Promise<ExecResult>;
|
|
update(id: string | number, values: Partial<T>): Promise<ExecResult>;
|
|
remove(id: string | number): Promise<ExecResult>;
|
|
exists(id: string | number): Promise<boolean>;
|
|
count(): Promise<number>;
|
|
}
|
|
|
|
export function createRepository<T extends Row>(
|
|
db: Db,
|
|
input: {
|
|
table: string;
|
|
idColumn?: string;
|
|
model?: Model<T>;
|
|
allowedColumns?: readonly (keyof T & string)[];
|
|
maxListLimit?: number;
|
|
/** Immutable equality scope (normally tenant_id) applied to every operation. */
|
|
scope?: { column: keyof T & string; value: unknown };
|
|
},
|
|
): Repository<T> {
|
|
const table = identifier(input.table);
|
|
const idColumn = identifier(input.idColumn ?? "id");
|
|
const allowed = input.allowedColumns ? new Set(input.allowedColumns.map(identifier)) : null;
|
|
const dialect = db.driver.dialect;
|
|
const maxListLimit = finiteInteger(input.maxListLimit, 1_000, 1);
|
|
const scopeColumn = input.scope ? identifier(input.scope.column) : null;
|
|
const valuesOf = (values: Partial<T>) => {
|
|
const entries = Object.entries(values).filter(([key, value]) => {
|
|
if (value === undefined || key === idColumn) return false;
|
|
identifier(key);
|
|
return !allowed || allowed.has(key) || key === scopeColumn;
|
|
});
|
|
if (!entries.length)
|
|
throw new TypeError("Repository write requires at least one allowed column.");
|
|
return entries;
|
|
};
|
|
const allowedReadColumn = (value: string): string => {
|
|
const safe = identifier(value);
|
|
if (allowed && safe !== idColumn && !allowed.has(safe)) {
|
|
throw new TypeError(`Repository column is not allowed: ${safe}`);
|
|
}
|
|
return safe;
|
|
};
|
|
|
|
return {
|
|
all: (options = {}) => {
|
|
const clauses: string[] = scopeColumn
|
|
? [`WHERE ${scopeColumn} = ${placeholder(dialect, 1)}`]
|
|
: [];
|
|
const params: unknown[] = scopeColumn ? [input.scope!.value] : [];
|
|
if (options.orderBy) {
|
|
clauses.push(
|
|
`ORDER BY ${allowedReadColumn(options.orderBy)} ${(options.direction ?? "asc").toUpperCase()}`,
|
|
);
|
|
}
|
|
if (options.limit !== undefined) {
|
|
const limit = Math.min(maxListLimit, finiteInteger(options.limit, maxListLimit, 1));
|
|
params.push(limit);
|
|
clauses.push(`LIMIT ${placeholder(dialect, params.length)}`);
|
|
}
|
|
if (options.offset !== undefined) {
|
|
const offset = finiteInteger(options.offset, 0, 0);
|
|
if (options.limit === undefined) {
|
|
params.push(maxListLimit);
|
|
clauses.push(`LIMIT ${placeholder(dialect, params.length)}`);
|
|
}
|
|
params.push(offset);
|
|
clauses.push(`OFFSET ${placeholder(dialect, params.length)}`);
|
|
}
|
|
return db.all<T>(
|
|
`SELECT * FROM ${table}${clauses.length ? ` ${clauses.join(" ")}` : ""}`,
|
|
params,
|
|
input.model,
|
|
);
|
|
},
|
|
find: (id) =>
|
|
db.one<T>(
|
|
`SELECT * FROM ${table} WHERE ${idColumn} = ${placeholder(dialect, 1)}${scopeColumn ? ` AND ${scopeColumn} = ${placeholder(dialect, 2)}` : ""} LIMIT 1`,
|
|
scopeColumn ? [id, input.scope!.value] : [id],
|
|
input.model,
|
|
),
|
|
require: (id) =>
|
|
firstOrThrow<T>(
|
|
db,
|
|
`SELECT * FROM ${table} WHERE ${idColumn} = ${placeholder(dialect, 1)}${scopeColumn ? ` AND ${scopeColumn} = ${placeholder(dialect, 2)}` : ""} LIMIT 1`,
|
|
scopeColumn ? [id, input.scope!.value] : [id],
|
|
input.model,
|
|
),
|
|
async create(values) {
|
|
const scoped = scopeColumn ? { ...values, [scopeColumn]: input.scope!.value } : values;
|
|
const entries = valuesOf(scoped);
|
|
return db.exec(
|
|
`INSERT INTO ${table} (${entries.map(([key]) => identifier(key)).join(", ")}) VALUES (${placeholders(dialect, entries.length).join(", ")})`,
|
|
entries.map(([, value]) => value),
|
|
);
|
|
},
|
|
async update(id, values) {
|
|
const entries = valuesOf(values);
|
|
return db.exec(
|
|
`UPDATE ${table} SET ${entries
|
|
.map(([key], index) => `${identifier(key)} = ${placeholder(dialect, index + 1)}`)
|
|
.join(
|
|
", ",
|
|
)} WHERE ${idColumn} = ${placeholder(dialect, entries.length + 1)}${scopeColumn ? ` AND ${scopeColumn} = ${placeholder(dialect, entries.length + 2)}` : ""}`,
|
|
[...entries.map(([, value]) => value), id, ...(scopeColumn ? [input.scope!.value] : [])],
|
|
);
|
|
},
|
|
remove: (id) =>
|
|
db.exec(
|
|
`DELETE FROM ${table} WHERE ${idColumn} = ${placeholder(dialect, 1)}${scopeColumn ? ` AND ${scopeColumn} = ${placeholder(dialect, 2)}` : ""}`,
|
|
scopeColumn ? [id, input.scope!.value] : [id],
|
|
),
|
|
exists: (id) =>
|
|
exists(
|
|
db,
|
|
`SELECT 1 FROM ${table} WHERE ${idColumn} = ${placeholder(dialect, 1)}${scopeColumn ? ` AND ${scopeColumn} = ${placeholder(dialect, 2)}` : ""} LIMIT 1`,
|
|
scopeColumn ? [id, input.scope!.value] : [id],
|
|
),
|
|
count: () =>
|
|
scopeColumn
|
|
? db
|
|
.one<{ count: number | string }>(
|
|
`SELECT COUNT(*) AS count FROM ${table} WHERE ${scopeColumn} = ${placeholder(dialect, 1)}`,
|
|
[input.scope!.value],
|
|
)
|
|
.then((row) => Number(row?.count ?? 0))
|
|
: countRows(db, table),
|
|
};
|
|
}
|