import "server-only"; import { execFile, spawn } from "node:child_process"; import { randomBytes } from "node:crypto"; import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import path from "node:path"; import { promisify } from "node:util"; import PocketBase, { ClientResponseError, isTokenExpired, type RecordModel } from "pocketbase"; import { DATA_ROOT } from "./paths"; export type { RecordModel }; /* * The data lives in PocketBase. Unless PB_URL points at a server of your own, Vectrum runs one * itself: bin/pocketbase (installed by scripts/install-pocketbase.mjs) with its data in * data/pb_data and the schema from pb_migrations/. The server talks to it as a superuser. */ const MANAGED = !process.env.PB_URL; export const PB_URL = (process.env.PB_URL ?? `http://127.0.0.1:${process.env.PB_PORT ?? 8090}`).replace(/\/+$/, ""); const BINARY = process.env.PB_BINARY ?? path.join(process.cwd(), "bin", process.platform === "win32" ? "pocketbase.exe" : "pocketbase"); const PB_DATA = path.join(DATA_ROOT, "pb_data"); const MIGRATIONS = path.join(process.cwd(), "pb_migrations"); type Credentials = { email: string; password: string }; /** From the environment, or (for the server Vectrum runs) generated once and kept in the data directory. */ function superuserCredentials(): Credentials { const { PB_SUPERUSER_EMAIL: email, PB_SUPERUSER_PASSWORD: password } = process.env; if (email && password) return { email, password }; if (!MANAGED) throw new Error("Set PB_SUPERUSER_EMAIL and PB_SUPERUSER_PASSWORD to use the PocketBase server at PB_URL."); const file = path.join(DATA_ROOT, "pocketbase-superuser.json"); if (!existsSync(file)) { mkdirSync(DATA_ROOT, { recursive: true }); const generated = { email: "superuser@vectrum.local", password: randomBytes(24).toString("base64url") }; writeFileSync(file, JSON.stringify(generated, null, 2), { mode: 0o600 }); } return JSON.parse(readFileSync(file, "utf8")) as Credentials; } const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); async function healthy() { try { return (await fetch(`${PB_URL}/api/health`, { signal: AbortSignal.timeout(1000) })).ok; } catch { return false; } } /** Starts PocketBase unless it already runs (e.g. left over from before a restart). */ async function ensureServer(credentials: Credentials) { if (await healthy()) return; if (!MANAGED) throw new Error(`PocketBase isn't reachable at ${PB_URL}.`); if (!existsSync(BINARY)) throw new Error(`PocketBase isn't installed at ${BINARY}. Run \`bun run pocketbase:install\`.`); const dirs = [`--dir=${PB_DATA}`, `--migrationsDir=${MIGRATIONS}`]; await promisify(execFile)(BINARY, ["superuser", "upsert", credentials.email, credentials.password, ...dirs]); const child = spawn(BINARY, ["serve", `--http=${new URL(PB_URL).host}`, ...dirs], { stdio: ["ignore", "ignore", "inherit"] }); process.once("exit", () => child.kill()); child.once("exit", (code) => console.error(`[pocketbase] exited with code ${code}`)); for (const deadline = Date.now() + 20_000; Date.now() < deadline; ) { if (await healthy()) return console.log(`[pocketbase] listening on ${PB_URL}`); if (child.exitCode !== null) break; await sleep(100); } throw new Error("PocketBase didn't start. See its output above."); } type Connection = { pb: PocketBase; credentials: Credentials; reauth?: Promise }; async function connect(): Promise { const credentials = superuserCredentials(); await ensureServer(credentials); const pb = client(); await pb.collection("_superusers").authWithPassword(credentials.email, credentials.password); return { pb, credentials }; } /** A client without credentials. Requests run concurrently, so the SDK mustn't cancel duplicates. */ export function client() { const pb = new PocketBase(PB_URL); pb.autoCancellation(false); return pb; } // One connection and one set of locks for the whole process: Next.js may load this module more than // once (route handlers, pages, instrumentation), and reloads it in development. const g = globalThis as unknown as { __pocketbase?: Promise; __dbLocks?: Map> }; /** The superuser client. */ export async function db() { const conn = await (g.__pocketbase ??= connect().catch((e) => { g.__pocketbase = undefined; throw e; })); if (isTokenExpired(conn.pb.authStore.token, 60)) { await (conn.reauth ??= conn.pb .collection("_superusers") .authWithPassword(conn.credentials.email, conn.credentials.password) .finally(() => (conn.reauth = undefined))); } return conn.pb; } /** A record id, for records created in batches that reference each other. */ export function newId() { const alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"; return Array.from(randomBytes(15), (b) => alphabet[b % alphabet.length]).join(""); } /** The first record matching a filter, or null. */ export async function findFirst(collection: string, filter: string, options: { sort?: string; expand?: string; fields?: string } = {}) { const { items } = await (await db()).collection(collection).getList(1, 1, { filter, skipTotal: true, ...options }); return items[0] ?? null; } /** A record by id, or null. */ export async function findById(collection: string, id: string, options: { expand?: string } = {}) { if (!id) return null; try { return await (await db()).collection(collection).getOne(id, options); } catch (e) { if (e instanceof ClientResponseError && e.status === 404) return null; throw e; } } /** How many records match a filter. */ export async function count(collection: string, filter: string) { return (await (await db()).collection(collection).getList(1, 1, { filter, fields: "id" })).totalItems; } /** Builds a filter expression with safely quoted parameters. */ export function filter(expression: string, params: Record = {}) { return client().filter(expression, params); } /** Whether a request failed because a value broke a unique index. */ export function isUniqueViolation(e: unknown) { if (!(e instanceof ClientResponseError) || e.status !== 400) return false; const data = (e.response?.data ?? {}) as Record; return Object.values(data).some((v) => v?.code === "validation_not_unique"); } const locks = (g.__dbLocks ??= new Map()); /** * Runs `fn` after every earlier call with the same key has finished. PocketBase has no * transactions over its API, so read-modify-write sequences (numbering issues, the job queue) * are serialized here. That holds because one Node.js process serves all requests. */ export function withLock(key: string, fn: () => Promise): Promise { const previous = locks.get(key) ?? Promise.resolve(); const result = previous.catch(() => {}).then(fn); const settled = result.catch(() => {}); locks.set(key, settled); settled.then(() => { if (locks.get(key) === settled) locks.delete(key); }); return result; } /** 0 and "" stand for "not set" in PocketBase number, text and relation fields. */ export const orNull = (value: T | null | undefined): T | null => (value ? value : null);