spb/market-atlas

1import { readdirSync, readFileSync, existsSync } from "node:fs";2import { dirname, join, resolve } from "node:path";3import { fileURLToPath } from "node:url";4import { pool } from "./pool.js";5import { logger } from "../logger.js";67/** Migrations live in infra/migrations (forward-only SQL files, applied in lexical order). */8export function migrationsDir(): string {9  const here = dirname(fileURLToPath(import.meta.url));10  const candidates = [11    resolve(here, "../../../../infra/migrations"), // apps/api/src/db → repo root12    resolve(here, "../../../infra/migrations"), // apps/api/dist → repo root13    resolve(process.cwd(), "infra/migrations"),14    resolve(process.cwd(), "../../infra/migrations"),15  ];16  const found = candidates.find((c) => existsSync(c));17  if (!found) throw new Error(`migrations directory not found (tried ${candidates.join(", ")})`);18  return found;19}2021export async function migrate(): Promise<string[]> {22  const dir = migrationsDir();23  const files = readdirSync(dir)24    .filter((f) => f.endsWith(".sql"))25    .sort();26  const client = await pool.connect();27  const applied: string[] = [];28  try {29    await client.query("create table if not exists schema_migrations (version text primary key, applied_at timestamptz not null default now())");30    await client.query("select pg_advisory_lock(7231)");31    const done = new Set((await client.query<{ version: string }>("select version from schema_migrations")).rows.map((r) => r.version));32    for (const f of files) {33      if (done.has(f)) continue;34      const sql = readFileSync(join(dir, f), "utf8");35      logger.info({ migration: f }, "applying migration");36      await client.query("begin");37      try {38        await client.query(sql);39        await client.query("insert into schema_migrations(version) values ($1)", [f]);40        await client.query("commit");41        applied.push(f);42      } catch (err) {43        await client.query("rollback");44        throw new Error(`migration ${f} failed: ${err instanceof Error ? err.message : String(err)}`);45      }46    }47    await client.query("select pg_advisory_unlock(7231)");48  } finally {49    client.release();50  }51  return applied;52}5354/** Ensure the daily partition for `day` (and the next one) exists. Idempotent. */55export async function ensureObservationPartition(dayMs: number): Promise<void> {56  const d = new Date(dayMs);57  d.setUTCHours(0, 0, 0, 0);58  for (const offset of [0, 1]) {59    const start = new Date(d.getTime() + offset * 86_400_000);60    const end = new Date(start.getTime() + 86_400_000);61    const name = `observations_${start.toISOString().slice(0, 10).replace(/-/g, "")}`;62    await pool.query(63      `create table if not exists ${name} partition of observations for values from ('${start.toISOString()}') to ('${end.toISOString()}')`,64    );65  }66}67