DenozordecandCursor
cff26813b9
Docker images / prepare-release (push) Successful in 9s
Docker images / backend-test (push) Successful in 2m18s
Docker images / frontend-image (push) Successful in 4m37s
Docker images / updater-image (push) Successful in 46s
Docker images / backend-image (push) Successful in 3m6s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 16s
Сохранять снимки в S3-compatible бакет, скачивать и удалять их вместе с локальными файлами. Привести /backups к DNA /servers: Frame, KPI, фильтры и реальное восстановление. Co-authored-by: Cursor <[email protected]>
537 lines
23 KiB
TypeScript
537 lines
23 KiB
TypeScript
import { createHash } from "node:crypto"
|
|
import { existsSync, readFileSync } from "node:fs"
|
|
import Database from "better-sqlite3"
|
|
import type { Pool } from "pg"
|
|
import { env } from "../config.js"
|
|
import { ensurePartitionsBetween, specForParent } from "./partitions.js"
|
|
import { encodeTrafficFlags } from "./traffic-flags.js"
|
|
|
|
export interface ImportReport {
|
|
sqlitePath: string
|
|
sqliteSha256: string
|
|
tables: Record<string, { sqlite: number; copied: number; skipped: number }>
|
|
rejects: string[]
|
|
durationMs: number
|
|
}
|
|
|
|
const SNAPSHOT_RETENTION_DAYS = 14
|
|
|
|
type ColKind = "ts" | "date" | "bool" | "json" | "json-null" | "bigint-id" | "int" | "text" | "num" | "flags" | "inet"
|
|
|
|
const INSERT_CHUNK = 1000
|
|
|
|
interface TableCopy {
|
|
table: string
|
|
columns: Array<[string, ColKind]>
|
|
timeCol?: string
|
|
retentionDays?: number
|
|
identity?: boolean
|
|
upsert?: boolean
|
|
}
|
|
|
|
const TABLES: TableCopy[] = [
|
|
{ table: "servers", identity: true, columns: [
|
|
["id", "int"], ["name", "text"], ["host", "text"], ["port", "int"], ["username", "text"],
|
|
["password", "text"], ["use_ssl", "bool"], ["verify_ssl", "bool"], ["type", "text"],
|
|
["site", "text"], ["country", "text"], ["asn", "text"], ["comment", "text"], ["enabled", "bool"],
|
|
["lan_subnet", "text"], ["wan_uplinks", "json"], ["mgmt_tunnel_ip", "text"],
|
|
["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "traffic_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["interval_sec", "int"], ["retention_days", "int"],
|
|
["last_collected_at", "ts"], ["last_duration_ms", "int"], ["last_error", "text"],
|
|
["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "servers_api_ping_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["interval_sec", "int"],
|
|
["last_collected_at", "ts"], ["last_duration_ms", "int"], ["last_error", "text"],
|
|
["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "traffic_flow_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["collector_ip", "text"], ["flow_listen_port", "int"],
|
|
["wg_listen_port", "int"], ["prefix", "text"], ["public_endpoint", "text"],
|
|
["host_public_key", "text"], ["host_private_key", "text"], ["hub_server_id", "int"],
|
|
["retention_hours", "int"], ["top_n", "int"], ["map_service_min_share_pct", "num"],
|
|
["last_datagram_at", "ts"], ["last_exporter_ip", "text"], ["last_error", "text"],
|
|
["packets_received", "int"], ["peers_json", "json"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "uptime_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["resources_enabled", "bool"], ["ping_enabled", "bool"],
|
|
["speed_enabled", "bool"], ["interval_sec", "int"], ["probe_interval_sec", "int"],
|
|
["speed_interval_sec", "int"], ["retention_days", "int"], ["last_collected_at", "ts"],
|
|
["last_duration_ms", "int"], ["last_error", "text"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "evobgp_settings", upsert: true, columns: [
|
|
["id", "int"], ["base_url", "text"], ["api_key", "text"], ["enabled", "bool"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_telegram_settings", upsert: true, columns: [
|
|
["id", "int"], ["bot_token", "text"], ["chat_id", "text"], ["message_thread_id", "int"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "acme_settings", upsert: true, columns: [
|
|
["id", "int"], ["directory_url", "text"], ["cloudflare_api_token", "text"],
|
|
["default_zone_id", "text"], ["account_private_key", "text"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "certificate_renew_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["interval_sec", "int"], ["renew_before_days", "int"],
|
|
["last_collected_at", "ts"], ["last_duration_ms", "int"], ["last_error", "text"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "backup_schedule_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["frequency", "text"], ["hour", "int"], ["minute", "int"],
|
|
["week_day", "int"], ["month_day", "int"], ["keep_count", "int"], ["format", "text"],
|
|
["server_ids_json", "json"], ["last_run_at", "ts"], ["last_duration_ms", "int"],
|
|
["last_error", "text"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "backup_storage_settings", upsert: true, columns: [
|
|
["id", "int"], ["provider", "text"], ["s3_endpoint", "text"], ["s3_region", "text"],
|
|
["s3_bucket", "text"], ["s3_prefix", "text"], ["s3_access_key_id", "text"],
|
|
["s3_secret_access_key", "text"], ["s3_force_path_style", "bool"], ["keep_local_copy", "bool"],
|
|
["last_test_at", "ts"], ["last_test_error", "text"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "internet_path_settings", upsert: true, columns: [
|
|
["id", "int"], ["enabled", "bool"], ["interval_sec", "int"], ["retention_days", "int"],
|
|
["last_collected_at", "ts"], ["last_duration_ms", "int"], ["last_error", "text"],
|
|
["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_engine_cursor", upsert: true, columns: [
|
|
["id", "int"], ["last_source_finished_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "filter_rules", identity: true, columns: [
|
|
["id", "int"], ["server_id", "int"], ["sort_order", "int"], ["community", "text"],
|
|
["community_name", "text"], ["action", "text"], ["gateway", "text"],
|
|
["gateway_tunnel_id", "text"], ["description", "text"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "recursive_routes", identity: true, columns: [
|
|
["id", "int"], ["server_id", "int"], ["sort_order", "int"], ["dst_address", "text"],
|
|
["gateway", "text"], ["distance", "int"], ["scope", "int"], ["target_scope", "int"],
|
|
["routing_table", "text"], ["check_gateway", "text"], ["country", "text"], ["comment", "text"],
|
|
["disabled", "bool"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "server_snapshots", identity: true, timeCol: "polled_at", retentionDays: SNAPSHOT_RETENTION_DAYS, columns: [
|
|
["id", "int"], ["server_id", "int"], ["polled_at", "ts"], ["status", "text"], ["latency_ms", "num"],
|
|
["ros_version", "text"], ["board_name", "text"], ["uptime", "text"], ["cpu_load", "int"],
|
|
["free_memory", "int"], ["total_memory", "int"], ["identity_name", "text"],
|
|
["raw_interfaces", "json-null"], ["raw_ip_addresses", "json-null"],
|
|
]},
|
|
{ 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"],
|
|
["flags", "flags"],
|
|
]},
|
|
{ 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", "inet"], ["dst", "inet"], ["proto", "int"],
|
|
["src_port", "int"], ["dst_port", "int"], ["bytes", "int"], ["packets", "int"],
|
|
["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: [
|
|
["server_id", "int"], ["bucket_at", "ts"], ["bytes", "int"], ["packets", "int"],
|
|
["unique_src", "int"], ["unique_dst", "int"], ["conversations", "int"],
|
|
]},
|
|
{ table: "flow_minute_dims", timeCol: "bucket_at", retentionDays: 3, columns: [
|
|
["server_id", "int"], ["bucket_at", "ts"], ["dim", "text"], ["key", "text"], ["bytes", "int"], ["packets", "int"],
|
|
]},
|
|
{ table: "flow_daily_dims", timeCol: "day", retentionDays: 396, columns: [
|
|
["server_id", "int"], ["day", "date"], ["dim", "text"], ["key", "text"], ["bytes", "int"], ["packets", "int"],
|
|
]},
|
|
{ table: "flow_ip_meta", columns: [
|
|
["prefix", "text"], ["asn", "int"], ["country", "text"], ["lat", "num"], ["lng", "num"],
|
|
["holder", "text"], ["ok", "int"], ["fetched_at", "ts"],
|
|
]},
|
|
{ table: "flow_asn_meta", columns: [
|
|
["asn", "int"], ["holder", "text"], ["fetched_at", "ts"],
|
|
]},
|
|
{ table: "uptime_probes", columns: [
|
|
["id", "text"], ["src_server_id", "int"], ["src_interface", "text"], ["name", "text"],
|
|
["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", timeCol: "sampled_at", retentionDays: 14, columns: [
|
|
["probe_id", "text"], ["sampled_at", "ts"], ["rtt_ms", "int"], ["loss_pct", "int"], ["status", "text"],
|
|
]},
|
|
{ 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"],
|
|
]},
|
|
{ table: "uptime_speed_probes", columns: [
|
|
["id", "text"], ["src_server_id", "int"], ["dst_server_id", "int"], ["src_interface", "text"],
|
|
["dst_interface", "text"], ["protocol", "text"], ["direction", "text"], ["duration_sec", "int"],
|
|
["enabled", "bool"], ["last_run_at", "ts"], ["last_tx_avg_mbps", "num"], ["last_rx_avg_mbps", "num"],
|
|
["last_status", "text"], ["last_error", "text"], ["last_ping_rtt_ms", "int"],
|
|
["last_ping_loss_pct", "int"], ["last_ping_at", "ts"], ["last_ping_error", "text"],
|
|
["sort_order", "int"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "uptime_speed_test_runs", columns: [
|
|
["id", "text"], ["probe_id", "text"], ["src_server_id", "int"], ["dst_server_id", "int"],
|
|
["src_interface", "text"], ["dst_interface", "text"], ["src_address", "text"], ["dst_address", "text"],
|
|
["src_interface_address", "text"], ["dst_interface_address", "text"], ["protocol", "text"],
|
|
["direction", "text"], ["duration_sec", "int"], ["tx_avg_mbps", "num"], ["rx_avg_mbps", "num"],
|
|
["ping_rtt_ms", "int"], ["ping_loss_pct", "int"], ["ping_error", "text"], ["status", "text"],
|
|
["error", "text"], ["created_at", "ts"],
|
|
]},
|
|
{ table: "app_users", columns: [
|
|
["id", "text"], ["name", "text"], ["login", "text"], ["email", "text"], ["role", "text"],
|
|
["active", "bool"], ["avatar", "text"], ["last_seen", "ts"], ["sections_json", "json"],
|
|
["servers_json", "json"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "user_interface_bindings", columns: [
|
|
["id", "text"], ["user_id", "text"], ["server_id", "int"], ["interface_name", "text"],
|
|
["interface_type", "text"], ["peer_public_key", "text"], ["peer_name", "text"],
|
|
["comment", "text"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_groups", columns: [
|
|
["id", "text"], ["name", "text"], ["combine_mode", "text"], ["enabled", "bool"],
|
|
["cooldown_override", "text"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_rules", columns: [
|
|
["id", "text"], ["name", "text"], ["type", "text"], ["target", "text"], ["condition", "text"],
|
|
["severity", "text"], ["enabled", "bool"], ["cooldown", "text"], ["rule_chat_id", "text"],
|
|
["confirm_stability_sec", "int"], ["recovery_mode", "text"], ["recovery_stability_sec", "int"],
|
|
["group_id", "text"], ["created_at", "ts"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_rule_targets", columns: [
|
|
["id", "text"], ["rule_id", "text"], ["target", "text"], ["sort_index", "int"],
|
|
]},
|
|
{ table: "alert_rule_conditions", columns: [
|
|
["id", "text"], ["rule_id", "text"], ["condition_line", "text"], ["sort_index", "int"],
|
|
]},
|
|
{ table: "alert_engine_state", columns: [
|
|
["scope_key", "text"], ["last_fired_at", "text"], ["last_payload_hash", "text"],
|
|
]},
|
|
{ table: "alert_engine_prev_live", columns: [
|
|
["kind", "text"], ["payload_json", "json"], ["updated_at", "ts"],
|
|
]},
|
|
{ table: "alert_engine_confirm_pending", columns: [
|
|
["rule_id", "text"], ["payload_hash", "text"], ["since_at", "ts"],
|
|
]},
|
|
{ table: "alert_history", columns: [
|
|
["id", "text"], ["rule_id", "text"], ["group_id", "text"], ["rule_name", "text"],
|
|
["severity", "text"], ["message", "text"], ["sent_ok", "bool"], ["fired_at", "ts"],
|
|
]},
|
|
{ table: "alert_outbox", columns: [
|
|
["id", "text"], ["dedupe_key", "text"], ["channel", "text"], ["status", "text"],
|
|
["retry_count", "int"], ["max_retries", "int"], ["next_attempt_at", "ts"],
|
|
["payload_json", "json"], ["created_at", "ts"], ["sent_at", "ts"], ["last_error", "text"],
|
|
]},
|
|
{ table: "certificate_issue_jobs", columns: [
|
|
["id", "text"], ["status", "text"], ["step", "text"], ["source", "text"],
|
|
["server_id", "bigint-id"], ["cert_name", "text"], ["domain_names", "json"],
|
|
["key_type", "text"], ["trust_store", "text"], ["requested_at", "ts"],
|
|
["started_at", "ts"], ["finished_at", "ts"], ["error", "text"],
|
|
]},
|
|
{ table: "backup_entries", columns: [
|
|
["id", "text"], ["server_id", "bigint-id"], ["server_name", "text"], ["filename", "text"],
|
|
["size_bytes", "int"], ["kind", "text"], ["notes", "text"], ["created_at", "ts"],
|
|
]},
|
|
{ table: "scheduler_runs", timeCol: "started_at", retentionDays: 30, columns: [
|
|
["id", "text"], ["job_key", "text"], ["started_at", "ts"], ["finished_at", "ts"],
|
|
["status", "text"], ["error", "text"], ["duration_ms", "int"], ["result_json", "json-null"],
|
|
]},
|
|
{ table: "events", timeCol: "created_at", retentionDays: 30, columns: [
|
|
["id", "text"], ["created_at", "ts"], ["level", "text"], ["event_type", "text"],
|
|
["source_module", "text"], ["title", "text"], ["message", "text"],
|
|
["entity_type", "text"], ["entity_id", "text"], ["payload_json", "json-null"],
|
|
]},
|
|
{ table: "internet_path_snapshots", identity: true, timeCol: "sampled_at", retentionDays: 14, columns: [
|
|
["id", "int"], ["sampled_at", "ts"], ["payload_json", "json"],
|
|
]},
|
|
]
|
|
|
|
function parseTs(value: unknown, strict: boolean, rejects: string[], ctx: string): string | null {
|
|
if (value == null || value === "") return null
|
|
const s = String(value)
|
|
try {
|
|
if (/^\d{4}-\d{2}-\d{2}T/.test(s)) return new Date(s).toISOString()
|
|
if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/.test(s)) {
|
|
return new Date(`${s.replace(" ", "T")}Z`).toISOString()
|
|
}
|
|
const d = new Date(s)
|
|
if (!Number.isNaN(d.getTime())) return d.toISOString()
|
|
} catch {
|
|
/* fallthrough */
|
|
}
|
|
const msg = `${ctx}: неразобранный timestamp ${s}`
|
|
if (strict) throw new Error(msg)
|
|
rejects.push(msg)
|
|
return null
|
|
}
|
|
|
|
function parseDate(value: unknown): string | null {
|
|
if (value == null || value === "") return null
|
|
return String(value).slice(0, 10)
|
|
}
|
|
|
|
function parseBool(value: unknown): boolean {
|
|
return value === true || value === 1 || value === "1" || value === "true"
|
|
}
|
|
|
|
function parseJson(value: unknown, fallback: unknown): unknown {
|
|
if (value == null || value === "") return fallback
|
|
if (typeof value !== "string") return value
|
|
try {
|
|
return JSON.parse(value) as unknown
|
|
} catch {
|
|
return fallback
|
|
}
|
|
}
|
|
|
|
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)
|
|
case "date":
|
|
return parseDate(value)
|
|
case "bool":
|
|
return parseBool(value)
|
|
case "json":
|
|
// Always JSON text. JS arrays must not go to node-pg as values — it encodes
|
|
// them as PG arrays (`{...}`), which jsonb rejects (22P02).
|
|
return JSON.stringify(parseJson(value, []))
|
|
case "json-null":
|
|
return value == null || value === "" ? null : JSON.stringify(parseJson(value, null))
|
|
case "bigint-id": {
|
|
const t = String(value ?? "").trim()
|
|
if (!t) return null
|
|
const n = Number(t)
|
|
if (!Number.isFinite(n)) {
|
|
const msg = `${ctx}: server_id ${t}`
|
|
if (strict) throw new Error(msg)
|
|
rejects.push(msg)
|
|
return null
|
|
}
|
|
return n
|
|
}
|
|
case "int":
|
|
if (value == null || value === "") return null
|
|
return Number(value)
|
|
case "num":
|
|
if (value == null || value === "") return null
|
|
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)
|
|
}
|
|
}
|
|
|
|
function sqliteTableExists(sqlite: Database.Database, name: string): boolean {
|
|
const row = sqlite.prepare(
|
|
`SELECT 1 AS ok FROM sqlite_master WHERE type = 'table' AND name = ?`,
|
|
).get(name) as { ok?: number } | undefined
|
|
return Boolean(row)
|
|
}
|
|
|
|
async function setval(pool: Pool, table: string): Promise<void> {
|
|
await pool.query(
|
|
`SELECT setval(
|
|
pg_get_serial_sequence($1, 'id'),
|
|
GREATEST(COALESCE((SELECT MAX(id) FROM ${table}), 1), 1),
|
|
true
|
|
)`,
|
|
[table],
|
|
)
|
|
}
|
|
|
|
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,
|
|
spec: TableCopy,
|
|
opts: { strict: boolean; fullHistory: boolean; rejects: string[] },
|
|
): Promise<{ sqlite: number; copied: number; skipped: number }> {
|
|
if (!sqliteTableExists(sqlite, spec.table)) {
|
|
return { sqlite: 0, copied: 0, skipped: 0 }
|
|
}
|
|
let where = ""
|
|
if (spec.timeCol && spec.retentionDays && !opts.fullHistory) {
|
|
const cutoff = new Date(Date.now() - spec.retentionDays * 86400_000).toISOString()
|
|
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
|
|
await precreatePartitions(sqlite, pool, spec, where)
|
|
const cols = spec.columns.map(([c]) => c)
|
|
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`
|
|
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")
|
|
await client.query(insertSql, batch.flat())
|
|
copied += batch.length
|
|
await client.query("COMMIT")
|
|
} catch (err) {
|
|
await client.query("ROLLBACK")
|
|
const detail = err instanceof Error ? err.message : String(err)
|
|
throw new Error(`${spec.table}: ${detail}`)
|
|
} finally {
|
|
client.release()
|
|
batch.length = 0
|
|
}
|
|
}
|
|
for (const row of stmt.iterate() as Iterable<Record<string, unknown>>) {
|
|
try {
|
|
const values = spec.columns.map(([col, kind]) =>
|
|
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 >= INSERT_CHUNK) await flush()
|
|
} catch (err) {
|
|
skipped += 1
|
|
const msg = `${spec.table}: ${err instanceof Error ? err.message : String(err)}`
|
|
opts.rejects.push(msg)
|
|
if (opts.strict) throw err
|
|
}
|
|
}
|
|
await flush()
|
|
if (spec.identity) {
|
|
try {
|
|
await setval(pool, spec.table)
|
|
} catch {
|
|
/* partitioned identity sequence name may differ */
|
|
}
|
|
}
|
|
return { sqlite: total, copied, skipped }
|
|
}
|
|
|
|
export function sqliteFileLooksPresent(path: string): boolean {
|
|
if (!existsSync(path) || path === ":memory:") return false
|
|
try {
|
|
const buf = readFileSync(path)
|
|
return buf.subarray(0, 16).toString("utf8").startsWith("SQLite format 3")
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
export async function shouldImportSqlite(pool: Pool, sqlitePath: string): Promise<boolean> {
|
|
if (!sqliteFileLooksPresent(sqlitePath)) return false
|
|
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) return false
|
|
const servers = await pool.query<{ c: string }>(`SELECT COUNT(*)::text AS c FROM servers`)
|
|
if (Number(servers.rows[0]?.c ?? 0) > 0) {
|
|
console.warn(
|
|
"SQLite → PostgreSQL: повтор недописанного импорта (маркера нет, servers уже не пустые)",
|
|
)
|
|
}
|
|
return true
|
|
}
|
|
|
|
export async function importSqliteToPostgres(
|
|
pool: Pool,
|
|
sqlitePath: string,
|
|
opts?: { dryRun?: boolean; strict?: boolean; fullHistory?: boolean },
|
|
): Promise<ImportReport> {
|
|
const started = Date.now()
|
|
const rejects: string[] = []
|
|
const sqlite = new Database(sqlitePath, { readonly: true, fileMustExist: true })
|
|
try {
|
|
sqlite.pragma("journal_mode = WAL")
|
|
} catch {
|
|
/* readonly */
|
|
}
|
|
const sha = createHash("sha256").update(readFileSync(sqlitePath)).digest("hex")
|
|
const report: ImportReport = {
|
|
sqlitePath,
|
|
sqliteSha256: sha,
|
|
tables: {},
|
|
rejects,
|
|
durationMs: 0,
|
|
}
|
|
const fullHistory = opts?.fullHistory ?? env.sqliteImportFullHistory
|
|
const strict = opts?.strict ?? true
|
|
if (opts?.dryRun) {
|
|
for (const spec of TABLES) {
|
|
if (!sqliteTableExists(sqlite, spec.table)) {
|
|
report.tables[spec.table] = { sqlite: 0, copied: 0, skipped: 0 }
|
|
continue
|
|
}
|
|
const c = (sqlite.prepare(`SELECT COUNT(*) AS c FROM ${spec.table}`).get() as { c: number }).c
|
|
report.tables[spec.table] = { sqlite: c, copied: 0, skipped: 0 }
|
|
}
|
|
sqlite.close()
|
|
report.durationMs = Date.now() - started
|
|
return report
|
|
}
|
|
for (const spec of TABLES) {
|
|
console.log(`SQLite → PostgreSQL: таблица ${spec.table}`)
|
|
report.tables[spec.table] = await copyTable(sqlite, pool, spec, { strict, fullHistory, rejects })
|
|
}
|
|
sqlite.close()
|
|
await pool.query(
|
|
`UPDATE data_migration
|
|
SET sqlite_imported_at = now(), sqlite_path = $1, sqlite_sha256 = $2, report_json = $3::jsonb
|
|
WHERE id = 1`,
|
|
[sqlitePath, sha, JSON.stringify(report.tables)],
|
|
)
|
|
report.durationMs = Date.now() - started
|
|
if (strict && rejects.length > 0) {
|
|
throw new Error(`SQLite import strict: ${rejects.length} rejects\n${rejects.slice(0, 20).join("\n")}`)
|
|
}
|
|
return report
|
|
}
|