perf(db): сжать схему PostgreSQL 18 и повторно загрузить SQLite

При первом рестарте wipe всех таблиц и импорт из SQLite в компактную схему. Retention через DROP PARTITION, lz4 и AIO worker.

Co-authored-by: Cursor <[email protected]>

backend/src/db/bootstrap.ts

@@ -1,7 +1,7 @@
 import { pool } from "./index.js"
 import { applySqlMigrations } from "./migrate.js"
 import { dropExpiredPartitions, ensurePartitionsAround } from "./partitions.js"
-import { importSqliteToPostgres, shouldImportSqlite } from "./sqlite-import.js"
+import { importSqliteToPostgres, shouldImportSqlite, sqliteFileLooksPresent } from "./sqlite-import.js"
 import { env } from "../config.js"
 
 const ETL_LOCK = 8723101
@@ -19,6 +19,15 @@ export async function initDatabase(): Promise<void> {
       console.log(
         `SQLite → PostgreSQL: готово за ${report.durationMs}ms, таблиц ${Object.keys(report.tables).length}`,
       )
+    } else {
+      const marker = await pool.query<{ sqlite_imported_at: string | null }>(
+        `SELECT sqlite_imported_at FROM data_migration WHERE id = 1`,
+      )
+      if (!marker.rows[0]?.sqlite_imported_at && !sqliteFileLooksPresent(env.DATABASE_PATH)) {
+        console.warn(
+          `SQLite → PostgreSQL: файл ${env.DATABASE_PATH} не найден, база после wipe остаётся пустой (defaults settings)`,
+        )
+      }
     }
   } finally {
     try {

backend/src/db/migrate.ts

@@ -1,9 +1,9 @@
-import { readFileSync } from "node:fs"
+import { readdirSync, readFileSync } from "node:fs"
 import { dirname, join } from "node:path"
 import { fileURLToPath } from "node:url"
 import type { Pool } from "pg"
 
-const MIGRATION_ID = "0000_postgresql"
+const FIRST_MIGRATION = "0000_postgresql.sql"
 
 function migrationsDir(): string {
   const here = dirname(fileURLToPath(import.meta.url))
@@ -14,7 +14,7 @@ function migrationsDir(): string {
   ]
   for (const dir of candidates) {
     try {
-      readFileSync(join(dir, `${MIGRATION_ID}.sql`), "utf8")
+      readFileSync(join(dir, FIRST_MIGRATION), "utf8")
       return dir
     } catch {
       /* try next */
@@ -23,6 +23,10 @@ function migrationsDir(): string {
   throw new Error("Не найден backend/drizzle/0000_postgresql.sql")
 }
 
+function migrationId(file: string): string {
+  return file.replace(/\.sql$/i, "")
+}
+
 export async function applySqlMigrations(pool: Pool): Promise<void> {
   await pool.query(`
     CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -30,14 +34,21 @@ export async function applySqlMigrations(pool: Pool): Promise<void> {
       applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
     )
   `)
-  const { rows } = await pool.query<{ id: string }>(
-    `SELECT id FROM schema_migrations WHERE id = $1`,
-    [MIGRATION_ID],
+  const dir = migrationsDir()
+  const files = readdirSync(dir)
+    .filter((f) => /^\d{4}_.+\.sql$/i.test(f))
+    .sort((a, b) => a.localeCompare(b))
+  const applied = new Set(
+    (await pool.query<{ id: string }>(`SELECT id FROM schema_migrations`)).rows.map((r) => r.id),
   )
-  if (rows.length > 0) return
-  const sql = readFileSync(join(migrationsDir(), `${MIGRATION_ID}.sql`), "utf8")
-  await pool.query(sql)
-  await pool.query(`INSERT INTO schema_migrations (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, [
-    MIGRATION_ID,
-  ])
+  for (const file of files) {
+    const id = migrationId(file)
+    if (applied.has(id)) continue
+    const sql = readFileSync(join(dir, file), "utf8")
+    await pool.query(sql)
+    await pool.query(
+      `INSERT INTO schema_migrations (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`,
+      [id],
+    )
+  }
 }

backend/src/db/partitions.ts

@@ -21,6 +21,37 @@ export const PARTITION_SPECS: PartitionSpec[] = [
   { parent: "internet_path_snapshots", kind: "week", keepDays: 21 },
 ]
 
+const WEEK_SLACK_DAYS = 7
+
+async function resolveKeepDays(pool: Pool): Promise<Map<string, number>> {
+  const map = new Map(PARTITION_SPECS.map((s) => [s.parent, s.keepDays]))
+  try {
+    const traffic = await pool.query<{ retention_days: number }>(
+      `SELECT retention_days FROM traffic_settings WHERE id = 1`,
+    )
+    const td = Number(traffic.rows[0]?.retention_days)
+    if (Number.isFinite(td) && td > 0) map.set("traffic_samples", td + WEEK_SLACK_DAYS)
+
+    const uptime = await pool.query<{ retention_days: number }>(
+      `SELECT retention_days FROM uptime_settings WHERE id = 1`,
+    )
+    const ud = Number(uptime.rows[0]?.retention_days)
+    if (Number.isFinite(ud) && ud > 0) {
+      map.set("uptime_probe_samples", ud + WEEK_SLACK_DAYS)
+      map.set("uptime_resource_samples", ud + WEEK_SLACK_DAYS)
+    }
+
+    const path = await pool.query<{ retention_days: number }>(
+      `SELECT retention_days FROM internet_path_settings WHERE id = 1`,
+    )
+    const pd = Number(path.rows[0]?.retention_days)
+    if (Number.isFinite(pd) && pd > 0) map.set("internet_path_snapshots", pd + WEEK_SLACK_DAYS)
+  } catch {
+    /* settings may be absent mid-migration */
+  }
+  return map
+}
+
 function utcDate(d: Date): Date {
   return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate()))
 }
@@ -82,10 +113,29 @@ export async function ensurePartitionFor(
   return name
 }
 
+export async function ensurePartitionsBetween(
+  pool: Pool,
+  parent: string,
+  kind: PartitionKind,
+  from: Date,
+  to: Date,
+): Promise<void> {
+  const start = from.getTime() <= to.getTime() ? from : to
+  const end = from.getTime() <= to.getTime() ? to : from
+  for (let t = new Date(start.getTime()); t <= end; ) {
+    await ensurePartitionFor(pool, parent, kind, t)
+    if (kind === "day") t = addUtcDays(t, 1)
+    else if (kind === "week") t = addUtcDays(t, 7)
+    else t = new Date(Date.UTC(t.getUTCFullYear(), t.getUTCMonth() + 1, 1))
+  }
+}
+
 export async function ensurePartitionsAround(pool: Pool, around = new Date()): Promise<void> {
+  const keepDays = await resolveKeepDays(pool)
   for (const spec of PARTITION_SPECS) {
+    const keep = keepDays.get(spec.parent) ?? spec.keepDays
     const daysAhead = spec.kind === "month" ? 40 : spec.kind === "week" ? 21 : 8
-    const start = addUtcDays(around, -spec.keepDays)
+    const start = addUtcDays(around, -keep)
     const end = addUtcDays(around, daysAhead)
     for (let t = new Date(start.getTime()); t < end; ) {
       await ensurePartitionFor(pool, spec.parent, spec.kind, t)
@@ -97,8 +147,10 @@ export async function ensurePartitionsAround(pool: Pool, around = new Date()): P
 }
 
 export async function dropExpiredPartitions(pool: Pool, around = new Date()): Promise<void> {
+  const keepDays = await resolveKeepDays(pool)
   for (const spec of PARTITION_SPECS) {
-    const cutoff = addUtcDays(around, -spec.keepDays)
+    const keep = keepDays.get(spec.parent) ?? spec.keepDays
+    const cutoff = addUtcDays(around, -keep)
     const { rows } = await pool.query<{ relname: string }>(
       `SELECT c.relname
        FROM pg_inherits i

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

@@ -1,6 +1,7 @@
 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())) {
@@ -80,4 +81,93 @@ if (!(await withPgOrSkip())) {
   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')
+  `)
+  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")
+}
+
+{
+  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 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")

backend/src/db/schema.ts

@@ -5,10 +5,12 @@ import {
   date,
   doublePrecision,
   index,
+  inet,
   integer,
   jsonb,
   pgTable,
   primaryKey,
+  smallint,
   text,
   timestamp,
   uniqueIndex,
@@ -135,15 +137,13 @@ export const serversApiPingSettings = pgTable("servers_api_ping_settings", {
 })
 
 export const serversRestPingSamples = pgTable("servers_rest_ping_samples", {
-  id:         idIdentity(),
   serverId:   intPkRef().references(() => servers.id, { onDelete: "cascade" }),
   sampledAt:  ts("sampled_at").notNull(),
   ok:         boolean("ok").notNull(),
   latencyMs:  integer("latency_ms"),
   error:      text("error"),
 }, (t) => [
-  primaryKey({ columns: [t.id, t.sampledAt] }),
-  index("idx_servers_rest_ping_samples_server_id").on(t.serverId, t.sampledAt),
+  primaryKey({ columns: [t.serverId, t.sampledAt] }),
 ])
 
 export const trafficFlowSettings = pgTable("traffic_flow_settings", {
@@ -211,16 +211,16 @@ export const flowDailyDims = pgTable("flow_daily_dims", {
 export const flowBuckets = pgTable("flow_buckets", {
   serverId:  intPkRef().references(() => servers.id, { onDelete: "cascade" }),
   bucketAt:  ts("bucket_at").notNull(),
-  src:       text("src").notNull(),
-  dst:       text("dst").notNull(),
-  proto:     integer("proto").notNull().default(0),
+  src:       inet("src").notNull(),
+  dst:       inet("dst").notNull(),
+  proto:     smallint("proto").notNull().default(0),
   srcPort:   integer("src_port").notNull().default(0),
   dstPort:   integer("dst_port").notNull().default(0),
   bytes:     bigint("bytes", { mode: "number" }).notNull().default(0),
   packets:   bigint("packets", { mode: "number" }).notNull().default(0),
   inIface:   text("in_iface").notNull().default(""),
   outIface:  text("out_iface").notNull().default(""),
-  nextHop:   text("next_hop").notNull().default(""),
+  nextHop:   inet("next_hop"),
   flowStartMs: bigint("flow_start_ms", { mode: "number" }).notNull().default(0),
   flowEndMs:   bigint("flow_end_ms", { mode: "number" }).notNull().default(0),
 }, (t) => [
@@ -249,7 +249,6 @@ export const flowAsnMeta = pgTable("flow_asn_meta", {
 })
 
 export const trafficSamples = pgTable("traffic_samples", {
-  id:            idIdentity(),
   serverId:      intPkRef().references(() => servers.id, { onDelete: "cascade" }),
   interfaceName: text("interface_name").notNull(),
   peerPublicKey: text("peer_public_key").notNull().default(""),
@@ -258,11 +257,9 @@ export const trafficSamples = pgTable("traffic_samples", {
   txBytes:       bigint("tx_bytes", { mode: "number" }).notNull().default(0),
   rxBps:         bigint("rx_bps", { mode: "number" }).notNull().default(0),
   txBps:         bigint("tx_bps", { mode: "number" }).notNull().default(0),
-  running:       boolean("running").notNull().default(false),
-  disabled:      boolean("disabled").notNull().default(false),
+  flags:         smallint("flags").notNull().default(0),
 }, (t) => [
-  primaryKey({ columns: [t.id, t.sampledAt] }),
-  index("idx_traffic_samples_server_time").on(t.serverId, t.sampledAt),
+  primaryKey({ columns: [t.serverId, t.sampledAt, t.interfaceName, t.peerPublicKey] }),
   index("idx_traffic_samples_server_iface_time").on(t.serverId, t.interfaceName, t.sampledAt),
 ])
 
@@ -302,19 +299,16 @@ export const uptimeProbes = pgTable("uptime_probes", {
 ])
 
 export const uptimeProbeSamples = pgTable("uptime_probe_samples", {
-  id:        idIdentity(),
   probeId:   text("probe_id").notNull().references(() => uptimeProbes.id, { onDelete: "cascade" }),
   sampledAt: ts("sampled_at").notNull(),
   rttMs:     integer("rtt_ms"),
   lossPct:   integer("loss_pct").notNull().default(0),
   status:    text("status", { enum: ["up", "warn", "down"] }).notNull().default("down"),
 }, (t) => [
-  primaryKey({ columns: [t.id, t.sampledAt] }),
-  index("idx_uptime_probe_samples_probe_time").on(t.probeId, t.sampledAt),
+  primaryKey({ columns: [t.probeId, t.sampledAt] }),
 ])
 
 export const uptimeResourceSamples = pgTable("uptime_resource_samples", {
-  id:            idIdentity(),
   serverId:      intPkRef().references(() => servers.id, { onDelete: "cascade" }),
   sampledAt:     ts("sampled_at").notNull(),
   status:        text("status", { enum: ["online", "offline"] }).notNull().default("offline"),
@@ -324,11 +318,8 @@ export const uptimeResourceSamples = pgTable("uptime_resource_samples", {
   freeHddSpace:  bigint("free_hdd_space", { mode: "number" }).notNull().default(0),
   totalHddSpace: bigint("total_hdd_space", { mode: "number" }).notNull().default(0),
   uptimeSeconds: bigint("uptime_seconds", { mode: "number" }).notNull().default(0),
-  boardName:     text("board_name").notNull().default(""),
-  rosVersion:    text("ros_version").notNull().default(""),
 }, (t) => [
-  primaryKey({ columns: [t.id, t.sampledAt] }),
-  index("idx_uptime_resource_samples_server_time").on(t.serverId, t.sampledAt),
+  primaryKey({ columns: [t.serverId, t.sampledAt] }),
 ])
 
 export const uptimeSpeedProbes = pgTable("uptime_speed_probes", {

backend/src/db/sqlite-import.ts

@@ -3,7 +3,8 @@ import { existsSync, readFileSync } from "node:fs"
 import Database from "better-sqlite3"
 import type { Pool } from "pg"
 import { env } from "../config.js"
-import { ensurePartitionFor, specForParent } from "./partitions.js"
+import { ensurePartitionsBetween, specForParent } from "./partitions.js"
+import { encodeTrafficFlags } from "./traffic-flags.js"
 
 export interface ImportReport {
   sqlitePath: string
@@ -15,7 +16,9 @@ export interface ImportReport {
 
 const SNAPSHOT_RETENTION_DAYS = 14
 
-type ColKind = "ts" | "date" | "bool" | "json" | "json-null" | "bigint-id" | "int" | "text" | "num"
+type ColKind = "ts" | "date" | "bool" | "json" | "json-null" | "bigint-id" | "int" | "text" | "num" | "flags" | "inet"
+
+const INSERT_CHUNK = 1000
 
 interface TableCopy {
   table: string
@@ -103,18 +106,18 @@ const TABLES: TableCopy[] = [
     ["free_memory", "int"], ["total_memory", "int"], ["identity_name", "text"],
     ["raw_interfaces", "json-null"], ["raw_ip_addresses", "json-null"],
   ]},
-  { table: "traffic_samples", identity: true, timeCol: "sampled_at", retentionDays: 14, columns: [
-    ["id", "int"], ["server_id", "int"], ["interface_name", "text"], ["peer_public_key", "text"],
+  { table: "traffic_samples", timeCol: "sampled_at", retentionDays: 14, columns: [
+    ["server_id", "int"], ["interface_name", "text"], ["peer_public_key", "text"],
     ["sampled_at", "ts"], ["rx_bytes", "int"], ["tx_bytes", "int"], ["rx_bps", "int"], ["tx_bps", "int"],
-    ["running", "bool"], ["disabled", "bool"],
+    ["flags", "flags"],
   ]},
-  { table: "servers_rest_ping_samples", identity: true, timeCol: "sampled_at", retentionDays: 30, columns: [
-    ["id", "int"], ["server_id", "int"], ["sampled_at", "ts"], ["ok", "bool"], ["latency_ms", "int"], ["error", "text"],
+  { table: "servers_rest_ping_samples", timeCol: "sampled_at", retentionDays: 30, columns: [
+    ["server_id", "int"], ["sampled_at", "ts"], ["ok", "bool"], ["latency_ms", "int"], ["error", "text"],
   ]},
   { table: "flow_buckets", timeCol: "bucket_at", retentionDays: 2, columns: [
-    ["server_id", "int"], ["bucket_at", "ts"], ["src", "text"], ["dst", "text"], ["proto", "int"],
+    ["server_id", "int"], ["bucket_at", "ts"], ["src", "inet"], ["dst", "inet"], ["proto", "int"],
     ["src_port", "int"], ["dst_port", "int"], ["bytes", "int"], ["packets", "int"],
-    ["in_iface", "text"], ["out_iface", "text"], ["next_hop", "text"],
+    ["in_iface", "text"], ["out_iface", "text"], ["next_hop", "inet"],
     ["flow_start_ms", "int"], ["flow_end_ms", "int"],
   ]},
   { table: "flow_minute_stats", timeCol: "bucket_at", retentionDays: 3, columns: [
@@ -139,13 +142,13 @@ const TABLES: TableCopy[] = [
     ["target", "text"], ["probe_filter", "text"], ["enabled", "bool"], ["interval_sec", "int"],
     ["show_on_dashboard", "bool"], ["sort_order", "int"], ["created_at", "ts"], ["updated_at", "ts"],
   ]},
-  { table: "uptime_probe_samples", identity: true, timeCol: "sampled_at", retentionDays: 14, columns: [
-    ["id", "int"], ["probe_id", "text"], ["sampled_at", "ts"], ["rtt_ms", "int"], ["loss_pct", "int"], ["status", "text"],
+  { table: "uptime_probe_samples", timeCol: "sampled_at", retentionDays: 14, columns: [
+    ["probe_id", "text"], ["sampled_at", "ts"], ["rtt_ms", "int"], ["loss_pct", "int"], ["status", "text"],
   ]},
-  { table: "uptime_resource_samples", identity: true, timeCol: "sampled_at", retentionDays: 14, columns: [
-    ["id", "int"], ["server_id", "int"], ["sampled_at", "ts"], ["status", "text"], ["cpu_load", "int"],
+  { table: "uptime_resource_samples", timeCol: "sampled_at", retentionDays: 14, columns: [
+    ["server_id", "int"], ["sampled_at", "ts"], ["status", "text"], ["cpu_load", "int"],
     ["free_memory", "int"], ["total_memory", "int"], ["free_hdd_space", "int"], ["total_hdd_space", "int"],
-    ["uptime_seconds", "int"], ["board_name", "text"], ["ros_version", "text"],
+    ["uptime_seconds", "int"],
   ]},
   { table: "uptime_speed_probes", columns: [
     ["id", "text"], ["src_server_id", "int"], ["dst_server_id", "int"], ["src_interface", "text"],
@@ -269,7 +272,20 @@ function parseJson(value: unknown, fallback: unknown): unknown {
   }
 }
 
-function coerce(kind: ColKind, value: unknown, strict: boolean, rejects: string[], ctx: string): unknown {
+function parseInet(value: unknown): string | null {
+  const s = String(value ?? "").trim()
+  if (!s) return null
+  return s
+}
+
+function coerce(
+  kind: ColKind,
+  value: unknown,
+  strict: boolean,
+  rejects: string[],
+  ctx: string,
+  row?: Record<string, unknown>,
+): unknown {
   switch (kind) {
     case "ts":
       return parseTs(value, strict, rejects, ctx)
@@ -303,6 +319,10 @@ function coerce(kind: ColKind, value: unknown, strict: boolean, rejects: string[
       return Number(value)
     case "text":
       return value == null ? "" : String(value)
+    case "flags":
+      return encodeTrafficFlags(parseBool(row?.running), parseBool(row?.disabled))
+    case "inet":
+      return parseInet(value)
     default:
       return value == null ? null : String(value)
   }
@@ -326,6 +346,35 @@ async function setval(pool: Pool, table: string): Promise<void> {
   )
 }
 
+function placeholderFor(kind: ColKind, index: number): string {
+  if (kind === "json" || kind === "json-null") return `$${index}::jsonb`
+  if (kind === "inet") return `$${index}::inet`
+  return `$${index}`
+}
+
+async function precreatePartitions(
+  sqlite: Database.Database,
+  pool: Pool,
+  spec: TableCopy,
+  where: string,
+): Promise<void> {
+  const part = specForParent(spec.table)
+  if (!part || !spec.timeCol) return
+  const bounds = sqlite.prepare(
+    `SELECT MIN(${spec.timeCol}) AS a, MAX(${spec.timeCol}) AS b FROM ${spec.table}${where}`,
+  ).get() as { a?: unknown; b?: unknown }
+  if (bounds?.a == null || bounds?.b == null) return
+  const isDate = spec.columns.find((c) => c[0] === spec.timeCol)?.[1] === "date"
+  const fromIso = isDate
+    ? `${String(bounds.a).slice(0, 10)}T00:00:00Z`
+    : parseTs(bounds.a, false, [], spec.table)
+  const toIso = isDate
+    ? `${String(bounds.b).slice(0, 10)}T00:00:00Z`
+    : parseTs(bounds.b, false, [], spec.table)
+  if (!fromIso || !toIso) return
+  await ensurePartitionsBetween(pool, spec.table, part.kind, new Date(fromIso), new Date(toIso))
+}
+
 async function copyTable(
   sqlite: Database.Database,
   pool: Pool,
@@ -341,28 +390,27 @@ async function copyTable(
     where = ` WHERE ${spec.timeCol} >= '${cutoff.replace("T", " ").slice(0, 19)}' OR ${spec.timeCol} >= '${cutoff}'`
   }
   const total = (sqlite.prepare(`SELECT COUNT(*) AS c FROM ${spec.table}${where}`).get() as { c: number }).c
-  const part = specForParent(spec.table)
+  await precreatePartitions(sqlite, pool, spec, where)
   const cols = spec.columns.map(([c]) => c)
-  const placeholders = spec.columns
-    .map(([, kind], i) => (kind === "json" || kind === "json-null" ? `$${i + 1}::jsonb` : `$${i + 1}`))
-    .join(", ")
   const conflictSql = spec.upsert
     ? `ON CONFLICT (id) DO UPDATE SET ${cols.filter((c) => c !== "id").map((c) => `${c} = EXCLUDED.${c}`).join(", ")}`
     : `ON CONFLICT DO NOTHING`
-  const insertSql = `INSERT INTO ${spec.table} (${cols.join(", ")}) VALUES (${placeholders}) ${conflictSql}`
   let copied = 0
   let skipped = 0
   const stmt = sqlite.prepare(`SELECT * FROM ${spec.table}${where}`)
   const batch: unknown[][] = []
   const flush = async () => {
     if (batch.length === 0) return
+    const n = spec.columns.length
+    const valuesSql = batch.map((_, i) =>
+      `(${spec.columns.map(([, kind], j) => placeholderFor(kind, i * n + j + 1)).join(", ")})`,
+    ).join(", ")
+    const insertSql = `INSERT INTO ${spec.table} (${cols.join(", ")}) VALUES ${valuesSql} ${conflictSql}`
     const client = await pool.connect()
     try {
       await client.query("BEGIN")
-      for (const values of batch) {
-        await client.query(insertSql, values)
-        copied += 1
-      }
+      await client.query(insertSql, batch.flat())
+      copied += batch.length
       await client.query("COMMIT")
     } catch (err) {
       await client.query("ROLLBACK")
@@ -375,22 +423,15 @@ async function copyTable(
   }
   for (const row of stmt.iterate() as Iterable<Record<string, unknown>>) {
     try {
-      if (part && spec.timeCol) {
-        const raw = row[spec.timeCol]
-        const ts = spec.columns.find((c) => c[0] === spec.timeCol)?.[1] === "date"
-          ? `${String(raw).slice(0, 10)}T00:00:00Z`
-          : parseTs(raw, false, opts.rejects, spec.table)
-        if (ts) await ensurePartitionFor(pool, spec.table, part.kind, new Date(ts))
-      }
       const values = spec.columns.map(([col, kind]) =>
-        coerce(kind, row[col], opts.strict, opts.rejects, `${spec.table}.${col}`),
+        coerce(kind, row[col], opts.strict, opts.rejects, `${spec.table}.${col}`, row),
       )
       if (spec.table === "certificate_issue_jobs" && values[4] == null) {
         skipped += 1
         continue
       }
       batch.push(values)
-      if (batch.length >= 200) await flush()
+      if (batch.length >= INSERT_CHUNK) await flush()
     } catch (err) {
       skipped += 1
       const msg = `${spec.table}: ${err instanceof Error ? err.message : String(err)}`

backend/src/db/traffic-flags.test.ts

@@ -0,0 +1,19 @@
+import assert from "node:assert/strict"
+import {
+  decodeTrafficSampleFlags,
+  encodeTrafficFlags,
+  trafficFlagDisabled,
+  trafficFlagRunning,
+} from "./traffic-flags.js"
+
+assert.equal(encodeTrafficFlags(true, false), 1)
+assert.equal(encodeTrafficFlags(false, true), 2)
+assert.equal(encodeTrafficFlags(true, true), 3)
+assert.equal(encodeTrafficFlags(false, false), 0)
+assert.equal(trafficFlagRunning(1), true)
+assert.equal(trafficFlagDisabled(1), false)
+assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false })
+assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false })
+assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false })
+
+console.log("traffic-flags.test.ts: ok")

backend/src/db/traffic-flags.ts

@@ -0,0 +1,24 @@
+export const TRAFFIC_FLAG_RUNNING = 1
+export const TRAFFIC_FLAG_DISABLED = 2
+
+export function encodeTrafficFlags(running: boolean, disabled: boolean): number {
+  return (running ? TRAFFIC_FLAG_RUNNING : 0) | (disabled ? TRAFFIC_FLAG_DISABLED : 0)
+}
+
+export function trafficFlagRunning(flags: number | null | undefined): boolean {
+  return ((Number(flags) || 0) & TRAFFIC_FLAG_RUNNING) !== 0
+}
+
+export function trafficFlagDisabled(flags: number | null | undefined): boolean {
+  return ((Number(flags) || 0) & TRAFFIC_FLAG_DISABLED) !== 0
+}
+
+export function decodeTrafficSampleFlags(flags: number | null | undefined): {
+  running: boolean
+  disabled: boolean
+} {
+  return {
+    running: trafficFlagRunning(flags),
+    disabled: trafficFlagDisabled(flags),
+  }
+}