MikrotikManager/backend/src/db/pg-schema.test.ts

import assert from "node:assert/strict"
import { withPgOrSkip } from "../test/pg.js"
import { dbQuery, pool } from "./index.js"
import { applySqlMigrations } from "./migrate.js"
import { ensurePartitionFor } from "./partitions.js"

if (!(await withPgOrSkip())) {
  console.log("pg-schema.test.ts: skip")
  process.exit(0)
}

await applySqlMigrations(pool)

{
  const { rows } = await dbQuery<{ n: string }>(`SELECT COUNT(*)::text AS n FROM servers`)
  assert.ok(rows[0])
}

{
  await dbQuery(`
    INSERT INTO servers (name, host) VALUES ('pg-schema-test', '127.0.0.1')
  `)
  const { rows } = await dbQuery<{ id: number }>(`SELECT id FROM servers WHERE name = 'pg-schema-test' LIMIT 1`)
  const id = rows[0]?.id
  assert.ok(id)

  await ensurePartitionFor(pool, "flow_buckets", "day", new Date())
  const bucketAt = new Date().toISOString()
  await dbQuery(`
    INSERT INTO flow_buckets (
      server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface
    ) VALUES ($1, $2, '10.0.0.1', '8.8.8.8', 6, 1, 443, 10, 1, 'wg-flow')
    ON CONFLICT DO NOTHING
  `, [id, bucketAt])

  await dbQuery(`DELETE FROM flow_buckets WHERE server_id = $1`, [id])
  await dbQuery(`DELETE FROM servers WHERE id = $1`, [id])
}

{
  await dbQuery(`
    INSERT INTO alert_outbox (id, dedupe_key, payload_json, next_attempt_at)
    VALUES ('pg-test-1', 'pg-dedupe-key', '{}'::jsonb, now())
    ON CONFLICT (dedupe_key) DO NOTHING
  `)
  let threw = false
  try {
    await dbQuery(`
      INSERT INTO alert_outbox (id, dedupe_key, payload_json, next_attempt_at)
      VALUES ('pg-test-2', 'pg-dedupe-key', '{}'::jsonb, now())
    `)
  } catch {
    threw = true
  }
  assert.equal(threw, true, "alert_outbox.dedupe_key UNIQUE")
  await dbQuery(`DELETE FROM alert_outbox WHERE dedupe_key = 'pg-dedupe-key'`)
}

{
  const peers = [{ endpoint: "msk-gw02.rtnt.top:13232", publicKey: "x" }]
  await dbQuery(
    `INSERT INTO alert_outbox (id, dedupe_key, payload_json, next_attempt_at)
     VALUES ('pg-json-arr', 'pg-json-arr', $1::jsonb, now())`,
    [JSON.stringify(peers)],
  )
  const { rows } = await dbQuery<{ payload_json: unknown }>(
    `SELECT payload_json FROM alert_outbox WHERE id = 'pg-json-arr'`,
  )
  assert.equal(Array.isArray(rows[0]?.payload_json), true)
  await dbQuery(`DELETE FROM alert_outbox WHERE id = 'pg-json-arr'`)

  let arrayAsPgArrayFailed = false
  try {
    await dbQuery(
      `INSERT INTO alert_outbox (id, dedupe_key, payload_json, next_attempt_at)
       VALUES ('pg-json-bad', 'pg-json-bad', $1, now())`,
      [peers],
    )
  } catch (err) {
    arrayAsPgArrayFailed = err instanceof Error && /json|22P02/i.test(err.message)
  }
  await dbQuery(`DELETE FROM alert_outbox WHERE id = 'pg-json-bad'`).catch(() => undefined)
  assert.equal(arrayAsPgArrayFailed, true, "JS array must not be bound as jsonb without stringify")
}

{
  const { rows } = await dbQuery<{ attname: string }>(`
    SELECT a.attname
    FROM pg_index i
    JOIN unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord) ON true
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum
    WHERE i.indrelid = 'traffic_samples'::regclass AND i.indisprimary
    ORDER BY k.ord
  `)
  assert.deepEqual(
    rows.map((r) => r.attname),
    ["server_id", "sampled_at", "interface_name", "peer_public_key"],
    "traffic_samples PK без id",
  )
}

{
  const { rows } = await dbQuery<{ column_name: string }>(`
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'traffic_samples'
  `)
  const cols = new Set(rows.map((r) => r.column_name))
  assert.equal(cols.has("id"), false)
  assert.equal(cols.has("running"), false)
  assert.equal(cols.has("disabled"), false)
  assert.equal(cols.has("flags"), true)
}

{
  const { rows } = await dbQuery<{ column_name: string; udt_name: string }>(`
    SELECT column_name, udt_name
    FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'flow_buckets'
      AND column_name IN ('src', 'dst', 'next_hop', 'proto', 'nat_src', 'nat_dst', 'nat_src_port', 'nat_dst_port')
  `)
  const by = Object.fromEntries(rows.map((r) => [r.column_name, r.udt_name]))
  assert.equal(by.src, "inet")
  assert.equal(by.dst, "inet")
  assert.equal(by.next_hop, "inet")
  assert.equal(by.proto, "int2")
  assert.equal(by.nat_src, "inet")
  assert.equal(by.nat_dst, "inet")
  assert.equal(by.nat_src_port, "int4")
  assert.equal(by.nat_dst_port, "int4")
}

{
  const { rows } = await dbQuery<{ indexdef: string }>(`
    SELECT indexdef FROM pg_indexes
    WHERE schemaname = 'public' AND tablename = 'traffic_samples'
  `)
  const defs = rows.map((r) => r.indexdef.toLowerCase())
  assert.equal(defs.some((d) => d.includes("using brin")), false, "нет BRIN на traffic_samples")
  assert.equal(
    defs.filter((d) => d.includes("idx_traffic_samples_server_iface_time")).length,
    1,
    "один btree (server_id, interface_name, sampled_at)",
  )
  assert.equal(defs.some((d) => d.includes("idx_traffic_samples_server_time")), false)
}

{
  const { rows } = await dbQuery<{ column_name: string }>(`
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'uptime_resource_samples'
  `)
  const cols = new Set(rows.map((r) => r.column_name))
  assert.equal(cols.has("board_name"), false)
  assert.equal(cols.has("ros_version"), false)
}

{
  const mig = await dbQuery<{ id: string }>(
    `SELECT id FROM schema_migrations WHERE id = '0002_compact_schema'`,
  )
  assert.equal(mig.rows.length, 1, "0002 применена")

  await dbQuery(`INSERT INTO servers (name, host) VALUES ('pg-wipe-idempotent', '127.0.0.1')`)
  await applySqlMigrations(pool)
  const still = await dbQuery<{ n: string }>(
    `SELECT COUNT(*)::text AS n FROM servers WHERE name = 'pg-wipe-idempotent'`,
  )
  assert.equal(still.rows[0]?.n, "1", "повторный applySqlMigrations не wipe")
  await dbQuery(`DELETE FROM servers WHERE name = 'pg-wipe-idempotent'`)
}

{
  const mig = await dbQuery<{ id: string }>(
    `SELECT id FROM schema_migrations WHERE id = '0006_config_revisions'`,
  )
  assert.equal(mig.rows.length, 1, "0006 применена")

  const { rows } = await dbQuery<{ column_name: string; udt_name: string }>(`
    SELECT column_name, udt_name FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'config_revisions'
  `)
  const by = Object.fromEntries(rows.map((r) => [r.column_name, r.udt_name]))
  assert.equal(by.payload, "jsonb")
  assert.equal(by.fingerprint, "text")
  assert.equal(by.section, "text")
}

{
  const mig = await dbQuery<{ id: string }>(
    `SELECT id FROM schema_migrations WHERE id = '0007_flow_buckets_nat'`,
  )
  assert.equal(mig.rows.length, 1, "0007 применена")
}

{
  const marker = await dbQuery<{ sqlite_imported_at: string | null }>(
    `SELECT sqlite_imported_at FROM data_migration WHERE id = 1`,
  )
  assert.ok(marker.rows[0], "data_migration singleton после wipe")
}

console.log("pg-schema.test.ts: ok")