sing-ui/lib/db.ts

import Database from "better-sqlite3";
import fs from "fs";
import path from "path";

export const DATA_DIR = path.join(process.cwd(), "data");
const DB_PATH = path.join(DATA_DIR, "sling-ui.db");

export interface Connection {
  id: number;
  name: string;
  type: string; // mysql | mariadb | tidb
  host: string;
  port: number;
  user: string;
  password: string;
  database: string;
  params: string; // JSON object: URL query params, e.g. {"tidb_skip_isolation_level_check":"1"}
  created_at: string;
}

export interface ConnectionInput {
  name: string;
  type: string;
  host: string;
  port: number;
  user: string;
  password: string;
  database: string;
  params?: string;
}

export interface StreamConfig {
  name: string;
  object?: string;
  primary_key?: string;
  update_key?: string;
  select?: string;
}

export interface Pipeline {
  id: number;
  name: string;
  source_conn_id: number;
  target_conn_id: number;
  schema_sync: number; // 1 | 0
  schema_scope: string; // 'selected' | 'all'
  include_fk: number; // 1 | 0
  mode: string;
  streams: string; // JSON: StreamConfig[]
  env: string; // JSON object: sling replication env, e.g. {"SLING_THREADS":"4"}
  created_at: string;
  updated_at: string;
}

export interface PipelineInput {
  name: string;
  source_conn_id: number;
  target_conn_id: number;
  schema_sync?: number;
  schema_scope?: string;
  include_fk?: number;
  mode?: string;
  streams?: string;
  env?: string;
}

export type RunStatus = "queued" | "running" | "success" | "failed" | "cancelled";

// Normalize a JSON-object column value from an API body: accept either a
// plain object or a pre-serialized JSON string.
export function normalizeJsonColumn(raw: unknown): string {
  if (typeof raw === "string") return raw;
  return JSON.stringify(raw ?? {});
}

export interface Run {
  id: number;
  pipeline_id: number;
  status: RunStatus;
  started_at: string | null;
  finished_at: string | null;
  log_path: string | null;
  error: string | null;
  pid: number | null;
  owner_pid: number | null;
}

function createDb(): Database.Database {
  fs.mkdirSync(DATA_DIR, { recursive: true });
  const db = new Database(DB_PATH);
  db.pragma("journal_mode = WAL");
  db.exec(`
    CREATE TABLE IF NOT EXISTS connections (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT UNIQUE NOT NULL,
      type TEXT NOT NULL,
      host TEXT NOT NULL,
      port INTEGER NOT NULL,
      user TEXT NOT NULL,
      password TEXT NOT NULL DEFAULT '',
      database TEXT NOT NULL,
      params TEXT NOT NULL DEFAULT '{}',
      created_at TEXT DEFAULT (datetime('now'))
    );
    CREATE TABLE IF NOT EXISTS pipelines (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT NOT NULL,
      source_conn_id INTEGER NOT NULL REFERENCES connections(id),
      target_conn_id INTEGER NOT NULL REFERENCES connections(id),
      schema_sync INTEGER DEFAULT 1,
      schema_scope TEXT NOT NULL DEFAULT 'selected',
      include_fk INTEGER NOT NULL DEFAULT 0,
      mode TEXT DEFAULT 'full-refresh',
      streams TEXT NOT NULL DEFAULT '[]',
      env TEXT NOT NULL DEFAULT '{}',
      created_at TEXT DEFAULT (datetime('now')),
      updated_at TEXT DEFAULT (datetime('now'))
    );
    CREATE TABLE IF NOT EXISTS runs (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      pipeline_id INTEGER NOT NULL REFERENCES pipelines(id),
      status TEXT NOT NULL DEFAULT 'queued',
      started_at TEXT,
      finished_at TEXT,
      log_path TEXT,
      error TEXT,
      pid INTEGER,
      owner_pid INTEGER
    );
  `);
  // Lightweight migrations for databases created before these columns existed.
  // Several Next.js workers may open the DB concurrently during build, so a
  // racing ALTER can fail with "duplicate column name" — safe to ignore.
  const addColumnIfMissing = (table: string, sql: string, column: string) => {
    const cols = new Set(
      (db.pragma(`table_info(${table})`) as { name: string }[]).map(
        (c) => c.name
      )
    );
    if (cols.has(column)) return;
    try {
      db.exec(sql);
    } catch (e) {
      if (!(e as Error).message.includes("duplicate column name")) throw e;
    }
  };
  addColumnIfMissing(
    "pipelines",
    "ALTER TABLE pipelines ADD COLUMN schema_scope TEXT NOT NULL DEFAULT 'selected'",
    "schema_scope"
  );
  addColumnIfMissing(
    "pipelines",
    "ALTER TABLE pipelines ADD COLUMN include_fk INTEGER NOT NULL DEFAULT 0",
    "include_fk"
  );
  addColumnIfMissing(
    "connections",
    "ALTER TABLE connections ADD COLUMN params TEXT NOT NULL DEFAULT '{}'",
    "params"
  );
  addColumnIfMissing(
    "pipelines",
    "ALTER TABLE pipelines ADD COLUMN env TEXT NOT NULL DEFAULT '{}'",
    "env"
  );
  addColumnIfMissing(
    "runs",
    "ALTER TABLE runs ADD COLUMN owner_pid INTEGER",
    "owner_pid"
  );
  return db;
}

// Keep a single connection across dev HMR reloads.
const globalForDb = globalThis as unknown as { __db?: Database.Database };
export const db: Database.Database =
  globalForDb.__db ?? (globalForDb.__db = createDb());

// ---------- connections ----------

export function listConnections(): Connection[] {
  return db
    .prepare("SELECT * FROM connections ORDER BY id")
    .all() as Connection[];
}

export function getConnection(id: number): Connection | null {
  return (
    (db.prepare("SELECT * FROM connections WHERE id = ?").get(id) as
      | Connection
      | undefined) ?? null
  );
}

export function createConnection(input: ConnectionInput): Connection {
  const res = db
    .prepare(
      `INSERT INTO connections (name, type, host, port, user, password, database, params)
       VALUES (@name, @type, @host, @port, @user, @password, @database, @params)`
    )
    .run({
      name: input.name,
      type: input.type,
      host: input.host,
      port: input.port,
      user: input.user,
      password: input.password ?? "",
      database: input.database,
      params: input.params ?? "{}",
    });
  return getConnection(Number(res.lastInsertRowid))!;
}

export function updateConnection(
  id: number,
  input: Partial<ConnectionInput>
): Connection | null {
  const fields = ["name", "type", "host", "port", "user", "password", "database", "params"].filter(
    (k) => input[k as keyof ConnectionInput] !== undefined
  );
  if (fields.length === 0) return getConnection(id);
  const sets = fields.map((f) => `${f} = @${f}`).join(", ");
  db.prepare(`UPDATE connections SET ${sets} WHERE id = @id`).run({
    ...input,
    id,
  });
  return getConnection(id);
}

export function deleteConnection(id: number): boolean {
  return db.prepare("DELETE FROM connections WHERE id = ?").run(id).changes > 0;
}

// ---------- pipelines ----------

export function listPipelines(): Pipeline[] {
  return db.prepare("SELECT * FROM pipelines ORDER BY id").all() as Pipeline[];
}

export function getPipeline(id: number): Pipeline | null {
  return (
    (db.prepare("SELECT * FROM pipelines WHERE id = ?").get(id) as
      | Pipeline
      | undefined) ?? null
  );
}

export function createPipeline(input: PipelineInput): Pipeline {
  const res = db
    .prepare(
      `INSERT INTO pipelines (name, source_conn_id, target_conn_id, schema_sync, schema_scope, include_fk, mode, streams, env)
       VALUES (@name, @source_conn_id, @target_conn_id, @schema_sync, @schema_scope, @include_fk, @mode, @streams, @env)`
    )
    .run({
      name: input.name,
      source_conn_id: input.source_conn_id,
      target_conn_id: input.target_conn_id,
      schema_sync: input.schema_sync ?? 1,
      schema_scope: input.schema_scope ?? "selected",
      include_fk: input.include_fk ?? 0,
      mode: input.mode ?? "full-refresh",
      streams: input.streams ?? "[]",
      env: input.env ?? "{}",
    });
  return getPipeline(Number(res.lastInsertRowid))!;
}

export function updatePipeline(
  id: number,
  input: Partial<PipelineInput>
): Pipeline | null {
  const fields = [
    "name",
    "source_conn_id",
    "target_conn_id",
    "schema_sync",
    "schema_scope",
    "include_fk",
    "mode",
    "streams",
    "env",
  ].filter((k) => input[k as keyof PipelineInput] !== undefined);
  if (fields.length === 0) return getPipeline(id);
  const sets = fields.map((f) => `${f} = @${f}`).join(", ");
  db.prepare(
    `UPDATE pipelines SET ${sets}, updated_at = datetime('now') WHERE id = @id`
  ).run({ ...input, id });
  return getPipeline(id);
}

export function deletePipeline(id: number): boolean {
  return db.prepare("DELETE FROM pipelines WHERE id = ?").run(id).changes > 0;
}

// ---------- runs ----------

export function createRun(input: {
  pipeline_id: number;
  status?: RunStatus;
  started_at?: string;
  log_path?: string;
  pid?: number;
  owner_pid?: number;
}): Run {
  const res = db
    .prepare(
      `INSERT INTO runs (pipeline_id, status, started_at, log_path, pid, owner_pid)
       VALUES (@pipeline_id, @status, @started_at, @log_path, @pid, @owner_pid)`
    )
    .run({
      pipeline_id: input.pipeline_id,
      status: input.status ?? "running",
      started_at: input.started_at ?? null,
      log_path: input.log_path ?? null,
      pid: input.pid ?? null,
      owner_pid: input.owner_pid ?? null,
    });
  return getRun(Number(res.lastInsertRowid))!;
}

export function getRun(id: number): Run | null {
  return (
    (db.prepare("SELECT * FROM runs WHERE id = ?").get(id) as Run | undefined) ??
    null
  );
}

export function updateRunStatus(
  id: number,
  patch: Partial<Pick<Run, "status" | "finished_at" | "error" | "pid" | "log_path">>
): Run | null {
  const fields = ["status", "finished_at", "error", "pid", "log_path"].filter(
    (k) => patch[k as keyof typeof patch] !== undefined
  );
  if (fields.length === 0) return getRun(id);
  const sets = fields.map((f) => `${f} = @${f}`).join(", ");
  db.prepare(`UPDATE runs SET ${sets} WHERE id = @id`).run({ ...patch, id });
  return getRun(id);
}

export interface RunWithPipeline extends Run {
  pipeline_name: string | null;
}

export function listRuns(): RunWithPipeline[] {
  return db
    .prepare(
      `SELECT r.*, p.name AS pipeline_name
       FROM runs r LEFT JOIN pipelines p ON p.id = r.pipeline_id
       ORDER BY r.id DESC`
    )
    .all() as RunWithPipeline[];
}