fix(netflow): починить live-аналитику и снизить запись в PostgreSQL

Коллектор снова отдаёт назначения в живом потоке. Ошибки записи видны в статусе, лишние перезаписи минутных агрегатов убраны.

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

.gitea/workflows/docker.yml

@@ -53,30 +53,39 @@ jobs:
         shell: bash
         run: |
           set -euo pipefail
-          docker run -d --name mm-pg \
+          NAME="mm-pg-${{ gitea.run_id }}-${{ gitea.run_attempt }}"
+          docker rm -f "$NAME" mm-pg 2>/dev/null || true
+          HOST_PORT=$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()')
+          docker run -d --name "$NAME" --rm \
             -e POSTGRES_USER=mmapp \
             -e POSTGRES_PASSWORD=mmapp \
             -e POSTGRES_DB=mmapp \
-            -p 5432:5432 \
+            -p "127.0.0.1:${HOST_PORT}:5432" \
             postgres:18-alpine
+          echo "PG_CONTAINER=$NAME" >> "${GITHUB_ENV}"
+          echo "DATABASE_URL=postgres://mmapp:[email protected]:${HOST_PORT}/mmapp" >> "${GITHUB_ENV}"
           for i in $(seq 1 40); do
-            if docker exec mm-pg pg_isready -U mmapp -d mmapp; then
+            if docker exec "$NAME" pg_isready -U mmapp -d mmapp; then
               exit 0
             fi
             sleep 1
           done
           echo "PostgreSQL не поднялся"
+          docker logs "$NAME" || true
           exit 1
 
       - name: Install and test backend
         shell: bash
-        env:
-          DATABASE_URL: postgres://mmapp:[email protected]:5432/mmapp
         run: |
           set -euo pipefail
           npm ci
           npm run test --prefix backend
 
+      - name: Stop PostgreSQL
+        if: always()
+        shell: bash
+        run: docker rm -f "${PG_CONTAINER:-}" mm-pg 2>/dev/null || true
+
   backend-image:
     needs:
       - prepare-release

README.md

@@ -128,7 +128,7 @@ sequenceDiagram
 | Зависимость | Реализация |
 |-------------|------------|
 | PostgreSQL | Контейнер `mmapp-postgres` (`postgres:18-alpine`). `DATABASE_URL=postgres://mmapp:…@postgres:5432/mmapp`. |
-| SQLite (ETL) | Файл `mikrotik.db` на томе `/app/data` (`DATABASE_PATH=/app/data/mikrotik.db`). При первом старте, если PG пустой, backend сам импортирует данные и ставит маркер. Повторный старт не копирует заново. |
+| SQLite (ETL) | Файл `mikrotik.db` на томе `/app/data` (`DATABASE_PATH=/app/data/mikrotik.db`). При старте, пока нет маркера `data_migration.sqlite_imported_at`, backend импортирует sqlite в PG (`ON CONFLICT` / upsert). После успешного импорта повтор не копирует заново. Том PG18: `mmapp-pgdata:/var/lib/postgresql` (не `.../data`). |
 | Docker socket | Только у контейнера updater: `/var/run/docker.sock` — доступ к Docker API хоста (управление контейнерами, pull). |
 
 ## Локальная разработка

app/(main)/traffic/page.tsx

@@ -952,10 +952,6 @@ export default function TrafficPage() {
       setFlowAnalytics(null)
       return
     }
-    if (range === "5m") {
-      setFlowAnalytics(null)
-      return
-    }
     if (range === "30d") {
       const month = new Date().toISOString().slice(0, 7)
       void getFlowMonthly(backendUrl, {
@@ -1128,7 +1124,7 @@ export default function TrafficPage() {
   const ingestLine = flowIngestLine(flowStats)
   const collectorAlive = Boolean(flowStats?.listenerBound || flowStats?.packetsReceived)
   const flowError = liveError
-    || (flowLiveError && !(collectorAlive && /live HTTP 500/.test(flowLiveError)) ? flowLiveError : null)
+    || flowLiveError
     || (displayedFlow?.degraded ? "Коллектор перегружен: упрощённая аналитика" : null)
 
   const flowKpiItems = [

backend/drizzle/0002_compact_schema.sql

@@ -0,0 +1,620 @@
+-- Compact PostgreSQL 18 schema: wipe all app tables (not schema_migrations),
+-- recreate with smaller time-series rows. SQLite is re-imported after this.
+
+DO $$
+DECLARE
+  r record;
+BEGIN
+  FOR r IN
+    SELECT tablename
+    FROM pg_tables
+    WHERE schemaname = 'public'
+      AND tablename <> 'schema_migrations'
+  LOOP
+    EXECUTE format('DROP TABLE IF EXISTS public.%I CASCADE', r.tablename);
+  END LOOP;
+END $$;
+
+CREATE TABLE IF NOT EXISTS schema_migrations (
+  id TEXT PRIMARY KEY,
+  applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS servers (
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+  name TEXT NOT NULL DEFAULT '',
+  host TEXT NOT NULL,
+  port INTEGER NOT NULL DEFAULT 443,
+  username TEXT NOT NULL DEFAULT 'admin',
+  password TEXT NOT NULL DEFAULT '',
+  use_ssl BOOLEAN NOT NULL DEFAULT TRUE,
+  verify_ssl BOOLEAN NOT NULL DEFAULT FALSE,
+  type TEXT NOT NULL DEFAULT 'home-router'
+    CHECK (type IN ('jump-host', 'exit-node', 'home-router')),
+  site TEXT NOT NULL DEFAULT '',
+  country TEXT NOT NULL DEFAULT '',
+  asn TEXT NOT NULL DEFAULT '',
+  comment TEXT NOT NULL DEFAULT '',
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  lan_subnet TEXT NOT NULL DEFAULT '',
+  wan_uplinks JSONB COMPRESSION lz4 NOT NULL DEFAULT '[]'::jsonb,
+  mgmt_tunnel_ip TEXT NOT NULL DEFAULT '',
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS traffic_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  interval_sec INTEGER NOT NULL DEFAULT 30,
+  retention_days INTEGER NOT NULL DEFAULT 14,
+  last_collected_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS servers_api_ping_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT FALSE,
+  interval_sec INTEGER NOT NULL DEFAULT 120,
+  last_collected_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS traffic_flow_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT FALSE,
+  collector_ip TEXT NOT NULL DEFAULT '10.255.254.1',
+  flow_listen_port INTEGER NOT NULL DEFAULT 4739,
+  wg_listen_port INTEGER NOT NULL DEFAULT 51821,
+  prefix TEXT NOT NULL DEFAULT '10.255.254.0/24',
+  public_endpoint TEXT NOT NULL DEFAULT '',
+  host_public_key TEXT NOT NULL DEFAULT '',
+  host_private_key TEXT NOT NULL DEFAULT '',
+  hub_server_id BIGINT,
+  retention_hours INTEGER NOT NULL DEFAULT 24,
+  top_n INTEGER NOT NULL DEFAULT 200,
+  map_service_min_share_pct DOUBLE PRECISION NOT NULL DEFAULT 5,
+  last_datagram_at TIMESTAMPTZ,
+  last_exporter_ip TEXT,
+  last_error TEXT,
+  packets_received BIGINT NOT NULL DEFAULT 0,
+  peers_json JSONB COMPRESSION lz4 NOT NULL DEFAULT '[]'::jsonb,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS uptime_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  resources_enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  ping_enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  speed_enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  interval_sec INTEGER NOT NULL DEFAULT 15,
+  probe_interval_sec INTEGER NOT NULL DEFAULT 15,
+  speed_interval_sec INTEGER NOT NULL DEFAULT 60,
+  retention_days INTEGER NOT NULL DEFAULT 14,
+  last_collected_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS evobgp_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  base_url TEXT NOT NULL DEFAULT '',
+  api_key TEXT NOT NULL DEFAULT '',
+  enabled BOOLEAN NOT NULL DEFAULT FALSE,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS alert_telegram_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  bot_token TEXT NOT NULL DEFAULT '',
+  chat_id TEXT NOT NULL DEFAULT '',
+  message_thread_id INTEGER,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS acme_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  directory_url TEXT NOT NULL DEFAULT 'https://acme-v02.api.letsencrypt.org/directory',
+  cloudflare_api_token TEXT NOT NULL DEFAULT '',
+  default_zone_id TEXT NOT NULL DEFAULT '',
+  account_private_key TEXT NOT NULL DEFAULT '',
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS certificate_renew_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  interval_sec INTEGER NOT NULL DEFAULT 21600,
+  renew_before_days INTEGER NOT NULL DEFAULT 30,
+  last_collected_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS backup_schedule_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  frequency TEXT NOT NULL DEFAULT 'daily' CHECK (frequency IN ('daily', 'weekly', 'monthly')),
+  hour INTEGER NOT NULL DEFAULT 3,
+  minute INTEGER NOT NULL DEFAULT 0,
+  week_day INTEGER NOT NULL DEFAULT 0,
+  month_day INTEGER NOT NULL DEFAULT 1,
+  keep_count INTEGER NOT NULL DEFAULT 7,
+  format TEXT NOT NULL DEFAULT 'rsc' CHECK (format IN ('rsc', 'backup')),
+  server_ids_json JSONB COMPRESSION lz4 NOT NULL DEFAULT '[]'::jsonb,
+  last_run_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS internet_path_settings (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  interval_sec INTEGER NOT NULL DEFAULT 300,
+  retention_days INTEGER NOT NULL DEFAULT 14,
+  last_collected_at TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS alert_engine_cursor (
+  id BIGINT PRIMARY KEY CHECK (id = 1),
+  last_source_finished_at TIMESTAMPTZ,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS data_migration (
+  id INTEGER PRIMARY KEY CHECK (id = 1),
+  sqlite_imported_at TIMESTAMPTZ,
+  sqlite_path TEXT,
+  sqlite_sha256 TEXT,
+  report_json JSONB COMPRESSION lz4
+);
+
+CREATE TABLE IF NOT EXISTS filter_rules (
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  sort_order INTEGER NOT NULL DEFAULT 0,
+  community TEXT NOT NULL,
+  community_name TEXT,
+  action TEXT NOT NULL DEFAULT 'route' CHECK (action IN ('route', 'blackhole')),
+  gateway TEXT NOT NULL DEFAULT '',
+  gateway_tunnel_id TEXT NOT NULL DEFAULT '',
+  description TEXT NOT NULL DEFAULT '',
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_filter_rules_server_sort ON filter_rules(server_id, sort_order);
+
+CREATE TABLE IF NOT EXISTS recursive_routes (
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  sort_order INTEGER NOT NULL DEFAULT 0,
+  dst_address TEXT NOT NULL,
+  gateway TEXT NOT NULL,
+  distance INTEGER NOT NULL DEFAULT 1,
+  scope INTEGER,
+  target_scope INTEGER,
+  routing_table TEXT NOT NULL DEFAULT 'main',
+  check_gateway TEXT NOT NULL DEFAULT '',
+  country TEXT NOT NULL DEFAULT '',
+  comment TEXT NOT NULL DEFAULT '',
+  disabled BOOLEAN NOT NULL DEFAULT FALSE,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_recursive_routes_server_sort ON recursive_routes(server_id, sort_order);
+
+CREATE TABLE IF NOT EXISTS server_snapshots (
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY,
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  polled_at TIMESTAMPTZ NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('online', 'offline')),
+  latency_ms DOUBLE PRECISION,
+  ros_version TEXT,
+  board_name TEXT,
+  uptime TEXT,
+  cpu_load INTEGER,
+  free_memory BIGINT,
+  total_memory BIGINT,
+  identity_name TEXT,
+  raw_interfaces JSONB COMPRESSION lz4,
+  raw_ip_addresses JSONB COMPRESSION lz4,
+  PRIMARY KEY (id, polled_at)
+) PARTITION BY RANGE (polled_at);
+CREATE INDEX IF NOT EXISTS idx_server_snapshots_server_time ON server_snapshots(server_id, polled_at);
+
+CREATE TABLE IF NOT EXISTS traffic_samples (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  interface_name TEXT NOT NULL,
+  peer_public_key TEXT NOT NULL DEFAULT '',
+  sampled_at TIMESTAMPTZ NOT NULL,
+  rx_bytes BIGINT NOT NULL DEFAULT 0,
+  tx_bytes BIGINT NOT NULL DEFAULT 0,
+  rx_bps BIGINT NOT NULL DEFAULT 0,
+  tx_bps BIGINT NOT NULL DEFAULT 0,
+  flags SMALLINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, sampled_at, interface_name, peer_public_key)
+) PARTITION BY RANGE (sampled_at);
+CREATE INDEX IF NOT EXISTS idx_traffic_samples_server_iface_time ON traffic_samples(server_id, interface_name, sampled_at);
+
+CREATE TABLE IF NOT EXISTS servers_rest_ping_samples (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  sampled_at TIMESTAMPTZ NOT NULL,
+  ok BOOLEAN NOT NULL,
+  latency_ms INTEGER,
+  error TEXT,
+  PRIMARY KEY (server_id, sampled_at)
+) PARTITION BY RANGE (sampled_at);
+
+CREATE TABLE IF NOT EXISTS flow_buckets (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  bucket_at TIMESTAMPTZ NOT NULL,
+  src INET NOT NULL,
+  dst INET NOT NULL,
+  proto SMALLINT NOT NULL DEFAULT 0,
+  src_port INTEGER NOT NULL DEFAULT 0,
+  dst_port INTEGER NOT NULL DEFAULT 0,
+  bytes BIGINT NOT NULL DEFAULT 0,
+  packets BIGINT NOT NULL DEFAULT 0,
+  in_iface TEXT NOT NULL DEFAULT '',
+  out_iface TEXT NOT NULL DEFAULT '',
+  next_hop INET,
+  flow_start_ms BIGINT NOT NULL DEFAULT 0,
+  flow_end_ms BIGINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
+) PARTITION BY RANGE (bucket_at);
+CREATE INDEX IF NOT EXISTS idx_flow_buckets_server_time ON flow_buckets(server_id, bucket_at);
+
+CREATE TABLE IF NOT EXISTS flow_minute_stats (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  bucket_at TIMESTAMPTZ NOT NULL,
+  bytes BIGINT NOT NULL DEFAULT 0,
+  packets BIGINT NOT NULL DEFAULT 0,
+  unique_src INTEGER NOT NULL DEFAULT 0,
+  unique_dst INTEGER NOT NULL DEFAULT 0,
+  conversations INTEGER NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, bucket_at)
+) PARTITION BY RANGE (bucket_at);
+
+CREATE TABLE IF NOT EXISTS flow_minute_dims (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  bucket_at TIMESTAMPTZ NOT NULL,
+  dim TEXT NOT NULL,
+  key TEXT NOT NULL,
+  bytes BIGINT NOT NULL DEFAULT 0,
+  packets BIGINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, bucket_at, dim, key)
+) PARTITION BY RANGE (bucket_at);
+CREATE INDEX IF NOT EXISTS idx_flow_minute_dims_time ON flow_minute_dims(bucket_at, dim);
+
+CREATE TABLE IF NOT EXISTS flow_daily_dims (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  day DATE NOT NULL,
+  dim TEXT NOT NULL,
+  key TEXT NOT NULL,
+  bytes BIGINT NOT NULL DEFAULT 0,
+  packets BIGINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, day, dim, key)
+) PARTITION BY RANGE (day);
+CREATE INDEX IF NOT EXISTS idx_flow_daily_dims_day ON flow_daily_dims(day, dim);
+
+CREATE TABLE IF NOT EXISTS flow_ip_meta (
+  prefix TEXT PRIMARY KEY,
+  asn INTEGER NOT NULL DEFAULT 0,
+  country TEXT NOT NULL DEFAULT '',
+  lat DOUBLE PRECISION,
+  lng DOUBLE PRECISION,
+  holder TEXT NOT NULL DEFAULT '',
+  ok INTEGER NOT NULL DEFAULT 1,
+  fetched_at TIMESTAMPTZ NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS flow_asn_meta (
+  asn INTEGER PRIMARY KEY,
+  holder TEXT NOT NULL DEFAULT '',
+  fetched_at TIMESTAMPTZ NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS uptime_probes (
+  id TEXT PRIMARY KEY,
+  src_server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  src_interface TEXT NOT NULL DEFAULT '',
+  name TEXT NOT NULL,
+  target TEXT NOT NULL,
+  probe_filter TEXT NOT NULL DEFAULT '—',
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  interval_sec INTEGER NOT NULL DEFAULT 0,
+  show_on_dashboard BOOLEAN NOT NULL DEFAULT FALSE,
+  sort_order INTEGER NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_uptime_probes_server_sort ON uptime_probes(src_server_id, sort_order);
+
+CREATE TABLE IF NOT EXISTS uptime_probe_samples (
+  probe_id TEXT NOT NULL REFERENCES uptime_probes(id) ON DELETE CASCADE,
+  sampled_at TIMESTAMPTZ NOT NULL,
+  rtt_ms INTEGER,
+  loss_pct INTEGER NOT NULL DEFAULT 0,
+  status TEXT NOT NULL DEFAULT 'down' CHECK (status IN ('up', 'warn', 'down')),
+  PRIMARY KEY (probe_id, sampled_at)
+) PARTITION BY RANGE (sampled_at);
+
+CREATE TABLE IF NOT EXISTS uptime_resource_samples (
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  sampled_at TIMESTAMPTZ NOT NULL,
+  status TEXT NOT NULL DEFAULT 'offline' CHECK (status IN ('online', 'offline')),
+  cpu_load INTEGER NOT NULL DEFAULT 0,
+  free_memory BIGINT NOT NULL DEFAULT 0,
+  total_memory BIGINT NOT NULL DEFAULT 0,
+  free_hdd_space BIGINT NOT NULL DEFAULT 0,
+  total_hdd_space BIGINT NOT NULL DEFAULT 0,
+  uptime_seconds BIGINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (server_id, sampled_at)
+) PARTITION BY RANGE (sampled_at);
+
+CREATE TABLE IF NOT EXISTS uptime_speed_probes (
+  id TEXT PRIMARY KEY,
+  src_server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  dst_server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  src_interface TEXT NOT NULL DEFAULT '',
+  dst_interface TEXT NOT NULL DEFAULT '',
+  protocol TEXT NOT NULL DEFAULT 'tcp' CHECK (protocol IN ('tcp', 'udp')),
+  direction TEXT NOT NULL DEFAULT 'both' CHECK (direction IN ('transmit', 'receive', 'both')),
+  duration_sec INTEGER NOT NULL DEFAULT 10,
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  last_run_at TIMESTAMPTZ,
+  last_tx_avg_mbps DOUBLE PRECISION,
+  last_rx_avg_mbps DOUBLE PRECISION,
+  last_status TEXT CHECK (last_status IN ('done', 'error')),
+  last_error TEXT,
+  last_ping_rtt_ms INTEGER,
+  last_ping_loss_pct INTEGER,
+  last_ping_at TIMESTAMPTZ,
+  last_ping_error TEXT,
+  sort_order INTEGER NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_uptime_speed_probes_src_sort ON uptime_speed_probes(src_server_id, sort_order);
+
+CREATE TABLE IF NOT EXISTS uptime_speed_test_runs (
+  id TEXT PRIMARY KEY,
+  probe_id TEXT,
+  src_server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  dst_server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  src_interface TEXT NOT NULL DEFAULT '',
+  dst_interface TEXT NOT NULL DEFAULT '',
+  src_address TEXT,
+  dst_address TEXT,
+  src_interface_address TEXT,
+  dst_interface_address TEXT,
+  protocol TEXT NOT NULL DEFAULT 'tcp' CHECK (protocol IN ('tcp', 'udp')),
+  direction TEXT NOT NULL DEFAULT 'both' CHECK (direction IN ('transmit', 'receive', 'both')),
+  duration_sec INTEGER NOT NULL DEFAULT 10,
+  tx_avg_mbps DOUBLE PRECISION,
+  rx_avg_mbps DOUBLE PRECISION,
+  ping_rtt_ms INTEGER,
+  ping_loss_pct INTEGER,
+  ping_error TEXT,
+  status TEXT NOT NULL DEFAULT 'done' CHECK (status IN ('done', 'error')),
+  error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_uptime_speed_test_runs_created_at ON uptime_speed_test_runs(created_at);
+
+CREATE TABLE IF NOT EXISTS certificate_issue_jobs (
+  id TEXT PRIMARY KEY,
+  status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'running', 'done', 'failed')),
+  step TEXT NOT NULL DEFAULT 'queued',
+  source TEXT NOT NULL DEFAULT 'manual' CHECK (source IN ('manual', 'scheduler')),
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  cert_name TEXT NOT NULL,
+  domain_names JSONB COMPRESSION lz4 NOT NULL,
+  key_type TEXT NOT NULL DEFAULT 'rsa2048',
+  trust_store TEXT NOT NULL DEFAULT 'www,api',
+  requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  started_at TIMESTAMPTZ,
+  finished_at TIMESTAMPTZ,
+  error TEXT
+);
+
+CREATE TABLE IF NOT EXISTS backup_entries (
+  id TEXT PRIMARY KEY,
+  server_id BIGINT REFERENCES servers(id) ON DELETE SET NULL,
+  server_name TEXT NOT NULL,
+  filename TEXT NOT NULL,
+  size_bytes BIGINT NOT NULL,
+  kind TEXT NOT NULL DEFAULT 'manual' CHECK (kind IN ('manual', 'auto')),
+  notes TEXT,
+  created_at TIMESTAMPTZ NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_backup_entries_filename ON backup_entries(filename);
+CREATE INDEX IF NOT EXISTS idx_backup_entries_server_created ON backup_entries(server_id, created_at);
+
+CREATE TABLE IF NOT EXISTS alert_groups (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL,
+  combine_mode TEXT NOT NULL DEFAULT 'any' CHECK (combine_mode IN ('any', 'all')),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  cooldown_override TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS alert_rules (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL,
+  type TEXT NOT NULL,
+  target TEXT NOT NULL,
+  condition TEXT NOT NULL,
+  severity TEXT NOT NULL CHECK (severity IN ('critical', 'warning', 'info')),
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  cooldown TEXT NOT NULL DEFAULT '5м',
+  rule_chat_id TEXT NOT NULL DEFAULT '',
+  confirm_stability_sec INTEGER,
+  recovery_mode TEXT NOT NULL DEFAULT 'always' CHECK (recovery_mode IN ('always', 'never', 'conditional')),
+  recovery_stability_sec INTEGER,
+  group_id TEXT REFERENCES alert_groups(id) ON DELETE SET NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS alert_rule_targets (
+  id TEXT PRIMARY KEY,
+  rule_id TEXT NOT NULL REFERENCES alert_rules(id) ON DELETE CASCADE,
+  target TEXT NOT NULL,
+  sort_index INTEGER NOT NULL DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_alert_rule_targets_rule ON alert_rule_targets(rule_id);
+
+CREATE TABLE IF NOT EXISTS alert_rule_conditions (
+  id TEXT PRIMARY KEY,
+  rule_id TEXT NOT NULL REFERENCES alert_rules(id) ON DELETE CASCADE,
+  condition_line TEXT NOT NULL,
+  sort_index INTEGER NOT NULL DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_alert_rule_conditions_rule ON alert_rule_conditions(rule_id);
+
+CREATE TABLE IF NOT EXISTS alert_engine_state (
+  scope_key TEXT PRIMARY KEY,
+  last_fired_at TEXT NOT NULL DEFAULT '',
+  last_payload_hash TEXT
+);
+
+CREATE TABLE IF NOT EXISTS alert_engine_prev_live (
+  kind TEXT PRIMARY KEY CHECK (kind IN ('gre', 'bgp')),
+  payload_json JSONB COMPRESSION lz4 NOT NULL DEFAULT '{}'::jsonb,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS alert_engine_confirm_pending (
+  rule_id TEXT PRIMARY KEY,
+  payload_hash TEXT NOT NULL,
+  since_at TIMESTAMPTZ NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS alert_history (
+  id TEXT PRIMARY KEY,
+  rule_id TEXT,
+  group_id TEXT,
+  rule_name TEXT NOT NULL,
+  severity TEXT NOT NULL CHECK (severity IN ('critical', 'warning', 'info')),
+  message TEXT NOT NULL,
+  sent_ok BOOLEAN NOT NULL DEFAULT TRUE,
+  fired_at TIMESTAMPTZ NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_alert_history_fired_at ON alert_history(fired_at);
+CREATE INDEX IF NOT EXISTS idx_alert_history_rule_id ON alert_history(rule_id);
+
+CREATE TABLE IF NOT EXISTS alert_outbox (
+  id TEXT PRIMARY KEY,
+  dedupe_key TEXT NOT NULL,
+  channel TEXT NOT NULL DEFAULT 'telegram' CHECK (channel IN ('telegram')),
+  status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
+  retry_count INTEGER NOT NULL DEFAULT 0,
+  max_retries INTEGER NOT NULL DEFAULT 3,
+  next_attempt_at TIMESTAMPTZ NOT NULL,
+  payload_json JSONB COMPRESSION lz4 NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  sent_at TIMESTAMPTZ,
+  last_error TEXT
+);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_alert_outbox_dedupe ON alert_outbox(dedupe_key);
+CREATE INDEX IF NOT EXISTS idx_alert_outbox_status_next_attempt ON alert_outbox(status, next_attempt_at);
+CREATE INDEX IF NOT EXISTS idx_alert_outbox_pending_next ON alert_outbox(next_attempt_at) WHERE status = 'pending';
+
+CREATE TABLE IF NOT EXISTS scheduler_runs (
+  id TEXT PRIMARY KEY,
+  job_key TEXT NOT NULL,
+  started_at TIMESTAMPTZ NOT NULL,
+  finished_at TIMESTAMPTZ NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('ok', 'error')),
+  error TEXT,
+  duration_ms INTEGER NOT NULL DEFAULT 0,
+  result_json JSONB COMPRESSION lz4
+);
+CREATE INDEX IF NOT EXISTS idx_scheduler_runs_job_time ON scheduler_runs(job_key, started_at);
+
+CREATE TABLE IF NOT EXISTS events (
+  id TEXT PRIMARY KEY,
+  created_at TIMESTAMPTZ NOT NULL,
+  level TEXT NOT NULL CHECK (level IN ('critical', 'warning', 'info')),
+  event_type TEXT NOT NULL,
+  source_module TEXT NOT NULL,
+  title TEXT NOT NULL,
+  message TEXT NOT NULL,
+  entity_type TEXT,
+  entity_id TEXT,
+  payload_json JSONB COMPRESSION lz4
+);
+CREATE INDEX IF NOT EXISTS idx_events_created_at ON events(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_level_created_at ON events(level, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_source_created_at ON events(source_module, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_event_type_created_at ON events(event_type, created_at DESC);
+
+CREATE TABLE IF NOT EXISTS app_users (
+  id TEXT PRIMARY KEY,
+  name TEXT NOT NULL DEFAULT '',
+  login TEXT NOT NULL UNIQUE,
+  email TEXT NOT NULL DEFAULT '',
+  role TEXT NOT NULL DEFAULT 'viewer' CHECK (role IN ('admin', 'operator', 'viewer')),
+  active BOOLEAN NOT NULL DEFAULT TRUE,
+  avatar TEXT NOT NULL DEFAULT '',
+  last_seen TIMESTAMPTZ,
+  sections_json JSONB COMPRESSION lz4 NOT NULL DEFAULT '[]'::jsonb,
+  servers_json JSONB COMPRESSION lz4 NOT NULL DEFAULT '[]'::jsonb,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS user_interface_bindings (
+  id TEXT PRIMARY KEY,
+  user_id TEXT NOT NULL REFERENCES app_users(id) ON DELETE CASCADE,
+  server_id BIGINT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
+  interface_name TEXT NOT NULL,
+  interface_type TEXT NOT NULL DEFAULT 'other' CHECK (interface_type IN ('ether', 'gre', 'wg', 'other')),
+  peer_public_key TEXT NOT NULL DEFAULT '',
+  peer_name TEXT NOT NULL DEFAULT '',
+  comment TEXT NOT NULL DEFAULT '',
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  UNIQUE (server_id, interface_name, peer_public_key)
+);
+CREATE INDEX IF NOT EXISTS idx_user_iface_bind_user ON user_interface_bindings(user_id);
+
+CREATE TABLE IF NOT EXISTS internet_path_snapshots (
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY,
+  sampled_at TIMESTAMPTZ NOT NULL,
+  payload_json JSONB COMPRESSION lz4 NOT NULL,
+  PRIMARY KEY (id, sampled_at)
+) PARTITION BY RANGE (sampled_at);
+CREATE INDEX IF NOT EXISTS idx_internet_path_snapshots_sampled ON internet_path_snapshots(sampled_at);
+
+INSERT INTO traffic_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO traffic_flow_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO uptime_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO evobgp_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO servers_api_ping_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO internet_path_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO alert_telegram_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO acme_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO certificate_renew_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO backup_schedule_settings (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO alert_engine_cursor (id) VALUES (1) ON CONFLICT (id) DO NOTHING;
+INSERT INTO data_migration (id) VALUES (1) ON CONFLICT (id) DO NOTHING;

backend/package.json

@@ -17,7 +17,7 @@
     "test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
     "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts",
     "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts",
-    "test:pg": "tsx src/db/sql-bind.test.ts && tsx src/db/pg-schema.test.ts",
+    "test:pg": "tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts",
     "test": "npm run test:alert-engine && npm run test:auth && npm run test:wireguard && npm run test:traffic-rate && npm run test:traffic-flow && npm run test:users && npm run test:pg"
   },
   "dependencies": {

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())) {
@@ -53,4 +54,120 @@ if (!(await withPgOrSkip())) {
   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')
+  `)
+  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)
@@ -278,9 +294,11 @@ function coerce(kind: ColKind, value: unknown, strict: boolean, rejects: string[
     case "bool":
       return parseBool(value)
     case "json":
-      return parseJson(value, [])
+      // 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 : parseJson(value, null)
+      return value == null || value === "" ? null : JSON.stringify(parseJson(value, null))
     case "bigint-id": {
       const t = String(value ?? "").trim()
       if (!t) return null
@@ -301,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)
   }
@@ -324,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,
@@ -339,30 +390,32 @@ 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 = cols.map((_, i) => `$${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")
-      throw err
+      const detail = err instanceof Error ? err.message : String(err)
+      throw new Error(`${spec.table}: ${detail}`)
     } finally {
       client.release()
       batch.length = 0
@@ -370,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)}`
@@ -421,7 +467,11 @@ export async function shouldImportSqlite(pool: Pool, sqlitePath: string): Promis
   )
   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) return false
+  if (Number(servers.rows[0]?.c ?? 0) > 0) {
+    console.warn(
+      "SQLite → PostgreSQL: повтор недописанного импорта (маркера нет, servers уже не пустые)",
+    )
+  }
   return true
 }
 
@@ -462,6 +512,7 @@ export async function importSqliteToPostgres(
     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()

backend/src/db/sqlite-json.test.ts

@@ -0,0 +1,11 @@
+import assert from "node:assert/strict"
+
+/** node-pg encodes a JS array as a PostgreSQL array (`{...}`), not JSON (`[...]`). */
+{
+  const peers = [{ endpoint: "msk-gw02.rtnt.top:13232" }]
+  const asJson = JSON.stringify(peers)
+  assert.equal(asJson.startsWith("["), true)
+  assert.equal(asJson.includes("msk-gw02.rtnt.top:13232"), true)
+}
+
+console.log("sqlite-json.test.ts: ok")

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),
+  }
+}

backend/src/modules/users/service/users-service.ts

@@ -13,6 +13,7 @@ import type {
 } from "@mmapp/contracts/users"
 import { db } from "../../../db/index.js"
 import { parseJsonArray } from "../../../db/json.js"
+import { decodeTrafficSampleFlags } from "../../../db/traffic-flags.js"
 import { servers, trafficSamples } from "../../../db/schema.js"
 import {
   createBindingRow,
@@ -270,8 +271,7 @@ export async function listInterfaceCatalog(serverId: number): Promise<CatalogInt
         .select({
           interfaceName: trafficSamples.interfaceName,
           peerPublicKey: trafficSamples.peerPublicKey,
-          running: trafficSamples.running,
-          disabled: trafficSamples.disabled,
+          flags: trafficSamples.flags,
         })
         .from(trafficSamples)
         .where(eq(trafficSamples.serverId, serverId)))
@@ -281,11 +281,12 @@ export async function listInterfaceCatalog(serverId: number): Promise<CatalogInt
       for (const r of rows) {
         if (seen.has(r.interfaceName)) continue
         seen.add(r.interfaceName)
+        const decoded = decodeTrafficSampleFlags(r.flags)
         ifaces.push({
           name: r.interfaceName,
           type: mapRosInterfaceType("", r.interfaceName),
-          running: Boolean(r.running),
-          disabled: Boolean(r.disabled),
+          running: decoded.running,
+          disabled: decoded.disabled,
         })
       }
     }

backend/src/routes/ospf.ts

@@ -217,7 +217,7 @@ async function getLatestSnapshotLatencyMs(serverId: number): Promise<number> {
   return latest.length > 0 ? Math.max(1, Math.round(latest[0].latencyMs ?? 100)) : 100
 }
 
-async function latestTrafficByInterface(serverId: number): Promise<Map<string, { id: number; serverId: number; disabled: boolean; sampledAt: string; interfaceName: string; peerPublicKey: string; rxBytes: number; txBytes: number; rxBps: number; txBps: number; running: boolean; }>> {
+async function latestTrafficByInterface(serverId: number): Promise<Map<string, TrafficSampleRow>> {
   const rows = await db
     .select()
     .from(trafficSamples)

backend/src/routes/traffic-flow.ts

@@ -275,7 +275,7 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
 
     try {
       while (!abort.signal.aborted) {
-        const payload = safeBuildLiveFlowSample(liveQuery)
+        const payload = await safeBuildLiveFlowSample(liveQuery)
         writeSse(reply.raw, payload.event, payload.data)
         await sleep(LIVE_TICK_MS, abort.signal)
       }

backend/src/routes/uptime.ts

@@ -1,7 +1,7 @@
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
 import { asc, desc, eq } from "drizzle-orm"
 import { db } from "../db/index.js"
-import { servers, uptimeSpeedProbes, uptimeSpeedTestRuns } from "../db/schema.js"
+import { servers, serverSnapshots, uptimeSpeedProbes, uptimeSpeedTestRuns } from "../db/schema.js"
 import { MikrotikClient } from "../services/mikrotik.js"
 import { scheduleAlertEngineAfterDataCollectors } from "../services/alert-collector-hooks.js"
 import { refreshScheduler, getSchedulerStatus } from "../services/scheduler.js"
@@ -502,7 +502,14 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
     })
 
     const resources = await Promise.all(allServers.map(async (s) => {
-      const rows = await readResourceSamplesSince(sinceIso, s.id)
+      const [rows, snap] = await Promise.all([
+        readResourceSamplesSince(sinceIso, s.id),
+        db.select({ boardName: serverSnapshots.boardName })
+          .from(serverSnapshots)
+          .where(eq(serverSnapshots.serverId, s.id))
+          .orderBy(desc(serverSnapshots.polledAt))
+          .limit(1),
+      ])
       const { row: pick, hasData } = pickResourceDisplayRow(rows, resourceFallbackMaxGapMs)
       const cpuHistory = toSeries(rows.map((r) => r.cpuLoad), 40)
       return {
@@ -515,7 +522,7 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
         hddUsed: Math.max(0, ((pick?.totalHddSpace ?? 0) - (pick?.freeHddSpace ?? 0)) / (1024 * 1024)),
         hddTotal: Math.max(0, (pick?.totalHddSpace ?? 0) / (1024 * 1024)),
         uptimeSeconds: pick?.uptimeSeconds ?? 0,
-        boardName: pick?.boardName || "RouterBOARD",
+        boardName: snap[0]?.boardName || "RouterBOARD",
         temp: undefined as number | undefined,
       }
     }))

backend/src/services/internet-path-collector.ts

@@ -1,4 +1,4 @@
-import { and, asc, desc, eq, lt } from "drizzle-orm"
+import { and, asc, desc, eq } from "drizzle-orm"
 import { db } from "../db/index.js"
 import {
   filterRules,
@@ -45,11 +45,6 @@ async function getSettingsRow() {
   return (await db.select().from(internetPathSettings).where(eq(internetPathSettings.id, 1)).limit(1))[0]
 }
 
-async function cleanupSnapshots(retentionDays: number) {
-  const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString()
-  await db.delete(internetPathSnapshots).where(lt(internetPathSnapshots.sampledAt, cutoff))
-}
-
 async function buildRulesets() {
   const enabled = await db.select().from(servers).where(eq(servers.enabled, true))
   const rules = await db.select().from(filterRules).orderBy(asc(filterRules.serverId), asc(filterRules.sortOrder))
@@ -273,7 +268,6 @@ export async function collectInternetPathSnapshotOnce(): Promise<InternetPathRun
       sampledAt,
       payloadJson: payload,
     })
-    await cleanupSnapshots(Math.max(1, settings.retentionDays))
     await db.update(internetPathSettings).set({
       lastCollectedAt: sampledAt,
       lastDurationMs: Date.now() - started,

backend/src/services/poller.ts

@@ -1,5 +1,5 @@
-import { eq, lt } from "drizzle-orm"
-import { db, pool } from "../db/index.js"
+import { eq } from "drizzle-orm"
+import { db, dbQuery, pool } from "../db/index.js"
 import { dropExpiredPartitions, ensurePartitionsAround } from "../db/partitions.js"
 import { servers, serverSnapshots } from "../db/schema.js"
 import type { SnapshotInsert } from "../db/schema.js"
@@ -82,11 +82,18 @@ export async function pollServer(serverId: number): Promise<SnapshotRead> {
     .values(partialSnap as SnapshotInsert)
     .returning()
 
-  const cutoffIso = new Date(Date.now() - 14 * 24 * 3600_000).toISOString()
-  await db.delete(serverSnapshots).where(lt(serverSnapshots.polledAt, cutoffIso))
+  if (inserted) {
+    await dbQuery(
+      `UPDATE server_snapshots
+       SET raw_interfaces = NULL, raw_ip_addresses = NULL
+       WHERE server_id = $1 AND polled_at < $2
+         AND (raw_interfaces IS NOT NULL OR raw_ip_addresses IS NOT NULL)`,
+      [serverId, inserted.polledAt],
+    )
+  }
   void dropExpiredPartitions(pool).then(() => ensurePartitionsAround(pool))
 
-  return toSnapshotRead(inserted)
+  return toSnapshotRead(inserted!)
 }
 
 // ── helper ─────────────────────────────────────────────────────────────────────

backend/src/services/servers-rest-ping-collector.ts

@@ -162,8 +162,6 @@ export async function collectServersRestPingOnce(): Promise<ServersRestPingRunSn
       }
     })
 
-    await dbQuery(`DELETE FROM servers_rest_ping_samples WHERE sampled_at < now() - interval '30 days'`)
-
     await db.update(serversApiPingSettings)
       .set({
         lastCollectedAt: sampledAt,

backend/src/services/traffic-collector.ts

@@ -1,6 +1,7 @@
-import { and, asc, desc, eq, gte, lt } from "drizzle-orm"
+import { and, asc, desc, eq, gte } from "drizzle-orm"
 import { db } from "../db/index.js"
 import { servers, trafficSamples, trafficSettings } from "../db/schema.js"
+import { decodeTrafficSampleFlags, encodeTrafficFlags } from "../db/traffic-flags.js"
 import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js"
 import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
 import { MikrotikClient } from "./mikrotik.js"
@@ -79,12 +80,6 @@ async function getSettingsRow() {
   return (await db.select().from(trafficSettings).where(eq(trafficSettings.id, 1)).limit(1))[0]
 }
 
-async function cleanupOldSamples(retentionDays: number) {
-  const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString()
-  await db.delete(trafficSamples)
-    .where(lt(trafficSamples.sampledAt, cutoff))
-}
-
 async function readPreviousWave(serverId: number): Promise<Map<string, { rxBytes: number; txBytes: number; sampledAt: string }>> {
   const last = (await db
     .select({ sampledAt: trafficSamples.sampledAt })
@@ -164,8 +159,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
             txBytes,
             rxBps,
             txBps,
-            running,
-            disabled,
+            flags: encodeTrafficFlags(running, disabled),
           }
         })
         try {
@@ -194,8 +188,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
               txBytes,
               rxBps,
               txBps,
-              running,
-              disabled,
+              flags: encodeTrafficFlags(running, disabled),
             })
           }
         } catch {
@@ -224,7 +217,6 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
       }
     }
 
-    await cleanupOldSamples(Math.max(1, settings.retentionDays))
     await db.update(trafficSettings).set({
       lastCollectedAt: now,
       lastDurationMs: Date.now() - startedAt,
@@ -286,11 +278,12 @@ export async function updateTrafficSettings(patch: {
 }
 
 export async function readServerSamplesInRange(serverId: number, sinceIso: string) {
-  return await db.select()
+  const rows = await db.select()
     .from(trafficSamples)
     .where(and(
       eq(trafficSamples.serverId, serverId),
       gte(trafficSamples.sampledAt, sinceIso),
     ))
     .orderBy(asc(trafficSamples.sampledAt))
+  return rows.map((r) => ({ ...r, ...decodeTrafficSampleFlags(r.flags) }))
 }

backend/src/services/traffic-flow-analytics.test.ts

@@ -16,6 +16,29 @@ import {
   seedRipeCacheForTests,
 } from "./traffic-flow-ripe.js"
 
+{
+  const liveErr = await formatLiveSseFromBuilder(() => {
+    throw new Error("SQLITE_BUSY")
+  })
+  assert.equal(liveErr.event, "error")
+  assert.equal((liveErr.data as { error: string }).error, "SQLITE_BUSY")
+  const liveOk = await formatLiveSseFromBuilder(() => ({ ok: true }))
+  assert.equal(liveOk.event, "sample")
+  const liveAsync = await formatLiveSseFromBuilder(async () => ({
+    uniqueSrc: 3,
+    destinations: [{ id: "8.8.8.8", label: "8.8.8.8", bytes: 1, packets: 1, bps: 1, percent: 100 }],
+  }))
+  assert.equal(liveAsync.event, "sample")
+  assert.notEqual(JSON.stringify(liveAsync.data), "{}")
+  assert.equal((liveAsync.data as { uniqueSrc: number }).uniqueSrc, 3)
+  assert.ok(Array.isArray((liveAsync.data as { destinations: unknown[] }).destinations))
+  const liveReject = await formatLiveSseFromBuilder(async () => {
+    throw new Error("pg down")
+  })
+  assert.equal(liveReject.event, "error")
+  assert.equal((liveReject.data as { error: string }).error, "pg down")
+}
+
 if (!(await withPgOrSkip())) {
   console.log("traffic-flow-analytics.test.ts: skip")
   process.exit(0)
@@ -248,13 +271,6 @@ try {
   assert.equal(degraded.degraded, true)
   assert.equal(degraded.conversationsList.length, 0)
   assert.ok((degraded.bytes ?? 0) >= 12_000)
-  const liveErr = formatLiveSseFromBuilder(() => {
-    throw new Error("SQLITE_BUSY")
-  })
-  assert.equal(liveErr.event, "error")
-  assert.equal((liveErr.data as { error: string }).error, "SQLITE_BUSY")
-  const liveOk = formatLiveSseFromBuilder(() => ({ ok: true }))
-  assert.equal(liveOk.event, "sample")
   const exporters = await listFlowExporters(5)
   const clients = await listFlowClients(5)
   assert.ok(Array.isArray(exporters.exporters))

backend/src/services/traffic-flow-analytics.ts

@@ -599,9 +599,12 @@ export async function listFlowClients(minutes: number): Promise<{ clients: { id:
   return { clients }
 }
 
-export function formatLiveSseFromBuilder(build: () => unknown): { event: "sample" | "error"; data: unknown } {
+export async function formatLiveSseFromBuilder(
+  build: () => unknown | Promise<unknown>,
+): Promise<{ event: "sample" | "error"; data: unknown }> {
   try {
-    return { event: "sample", data: build() }
+    const data = await build()
+    return { event: "sample", data }
   } catch (err) {
     const message = err instanceof Error ? err.message : String(err)
     return { event: "error", data: { error: message } }
@@ -613,10 +616,10 @@ export function isFlowAnalyticsDegraded(): boolean {
   return health.pendingSize >= LIVE_DEGRADED_PENDING
 }
 
-export function safeBuildLiveFlowSample(q: Omit<FlowAnalyticsQuery, "minutes" | "skipHeavy">): {
+export async function safeBuildLiveFlowSample(q: Omit<FlowAnalyticsQuery, "minutes" | "skipHeavy">): Promise<{
   event: "sample" | "error"
   data: unknown
-} {
+}> {
   return formatLiveSseFromBuilder(async () => {
     const skipHeavy = isFlowAnalyticsDegraded()
     return await buildFlowAnalytics({ ...q, minutes: LIVE_ANALYTICS_MINUTES, skipHeavy })

backend/src/services/traffic-flow-collector-worker.ts

@@ -60,7 +60,7 @@ function stopListener(): void {
     clearInterval(flushTimer)
     flushTimer = null
   }
-  void flushPending().catch((e) => {
+  void flushPending({ force: true }).catch((e) => {
     setEngineError(e instanceof Error ? e.message : String(e))
   })
   if (socket) {

backend/src/services/traffic-flow-engine.ts

@@ -12,6 +12,8 @@ import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
 import { pickInternetPeer } from "./traffic-flow-ip.js"
 
 export const TICK_MS = 2_000
+export const PERSIST_MS = 10_000
+export const STATS_PERSIST_MS = 15_000
 export const RING_LEN = 60
 export const MAX_PENDING = 50_000
 export const DAILY_ASN_TOP = 500
@@ -48,6 +50,66 @@ export interface PendingFlowRow {
   flowEndMs: number
 }
 
+function inetOrNull(value: string | null | undefined): string | null {
+  const s = String(value ?? "").trim()
+  return s.length > 0 ? s : null
+}
+
+export function isValidFlowInet(value: string): boolean {
+  const s = value.trim()
+  if (!s) return false
+  const v4 = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s)
+  if (v4) {
+    return v4.slice(1).every((octet) => {
+      const n = Number(octet)
+      return Number.isInteger(n) && n >= 0 && n <= 255
+    })
+  }
+  if (!s.includes(":")) return false
+  if (!/^[0-9a-fA-F:]+$/.test(s)) return false
+  const parts = s.split(":")
+  if (parts.length < 3 || parts.length > 8) return false
+  return parts.every((p) => p.length <= 4)
+}
+
+function clampProto(n: number): number {
+  if (!Number.isFinite(n)) return 0
+  return Math.max(0, Math.min(255, Math.trunc(n)))
+}
+
+function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
+  const src = (r.src || "").trim() || "0.0.0.0"
+  const dst = (r.dst || "").trim() || "0.0.0.0"
+  if (!isValidFlowInet(src) || !isValidFlowInet(dst)) return null
+  const next = inetOrNull(r.nextHop)
+  return {
+    ...r,
+    src,
+    dst,
+    nextHop: next && isValidFlowInet(next) ? next : "",
+    proto: clampProto(r.proto),
+  }
+}
+
+function flowUpsertParams(r: PendingFlowRow) {
+  return {
+    serverId: r.serverId,
+    bucketAt: r.bucketAt,
+    src: r.src,
+    dst: r.dst,
+    proto: r.proto,
+    srcPort: r.srcPort,
+    dstPort: r.dstPort,
+    bytes: r.bytes,
+    packets: r.packets,
+    inIface: r.inIface,
+    outIface: r.outIface,
+    nextHop: inetOrNull(r.nextHop),
+    flowStartMs: r.flowStartMs,
+    flowEndMs: r.flowEndMs,
+  }
+}
+
 export interface EngineStats {
   packetsReceived: number
   lastExporterIp: string | null
@@ -105,6 +167,9 @@ let rowsStored = 0
 let lastFlushUsedTransaction = false
 let lastPruneAt = 0
 let lastPassiveCheckpointAt = 0
+let lastPersistAt = 0
+let lastFlushedMinute = ""
+let lastStatsPersistAt = 0
 let dataEpoch = 0
 let lastPersistedStats: {
   packetsReceived: number
@@ -449,7 +514,7 @@ export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; ou
   }
 }
 
-async function persistListenerStats(): Promise<boolean> {
+async function persistListenerStats(force = false): Promise<boolean> {
   if (
     lastPersistedStats
     && lastPersistedStats.packetsReceived === packetsReceived
@@ -459,6 +524,12 @@ async function persistListenerStats(): Promise<boolean> {
   ) {
     return false
   }
+  const errorChanged = lastPersistedStats?.lastError !== lastError
+  const exporterChanged = lastPersistedStats?.lastExporterIp !== lastExporterIp
+  const now = Date.now()
+  if (!force && !errorChanged && !exporterChanged && now - lastStatsPersistAt < STATS_PERSIST_MS) {
+    return false
+  }
   await dbQuery(`
     UPDATE traffic_flow_settings
     SET packets_received = @packetsReceived,
@@ -480,6 +551,7 @@ async function persistListenerStats(): Promise<boolean> {
     lastExporterIp,
     lastError,
   }
+  lastStatsPersistAt = now
   invalidateTrafficFlowSettingsCache()
   return true
 }
@@ -644,20 +716,70 @@ function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
   return out
 }
 
-export async function flushPending(): Promise<void> {
-  pruneRecent()
-  rollFlowRings()
-  await persistListenerStats()
-  if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) {
-    await pruneStored()
-    lastFlushUsedTransaction = false
-    return
-  }
-  const rows = topNPending([...pending.values()].map(toPendingRow))
-  pending.clear()
-  for (const row of rows) mergeInto(recent, row)
+function persistDue(force: boolean, hasWork: boolean): boolean {
+  if (force) return true
+  if (!hasWork) return false
+  if (minuteBucketIso() !== lastFlushedMinute) return true
+  return Date.now() - lastPersistAt >= PERSIST_MS
+}
 
-  const upsertSql = `
+async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
+  if (rows.length === 0) return
+  const days = new Set(rows.map((r) => r.bucketAt))
+  for (const bucketAt of days) await ensureParentPartition("flow_buckets", bucketAt)
+  await pool.query({
+    text: `
+      INSERT INTO flow_buckets (
+        server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
+      )
+      SELECT *
+      FROM UNNEST(
+        $1::bigint[],
+        $2::timestamptz[],
+        $3::inet[],
+        $4::inet[],
+        $5::smallint[],
+        $6::int[],
+        $7::int[],
+        $8::bigint[],
+        $9::bigint[],
+        $10::text[],
+        $11::text[],
+        $12::inet[],
+        $13::bigint[],
+        $14::bigint[]
+      ) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms)
+      ON CONFLICT (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
+      DO UPDATE SET
+        bytes = flow_buckets.bytes + excluded.bytes,
+        packets = flow_buckets.packets + excluded.packets,
+        out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
+        next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
+        flow_start_ms = CASE
+          WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
+          THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
+        flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
+    `,
+    values: [
+      rows.map((r) => r.serverId),
+      rows.map((r) => r.bucketAt),
+      rows.map((r) => r.src),
+      rows.map((r) => r.dst),
+      rows.map((r) => r.proto),
+      rows.map((r) => r.srcPort),
+      rows.map((r) => r.dstPort),
+      rows.map((r) => r.bytes),
+      rows.map((r) => r.packets),
+      rows.map((r) => r.inIface),
+      rows.map((r) => r.outIface),
+      rows.map((r) => inetOrNull(r.nextHop)),
+      rows.map((r) => r.flowStartMs),
+      rows.map((r) => r.flowEndMs),
+    ],
+  })
+}
+
+const FLOW_UPSERT_SQL = `
     INSERT INTO flow_buckets (
       server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
     ) VALUES (
@@ -668,61 +790,93 @@ export async function flushPending(): Promise<void> {
       bytes = flow_buckets.bytes + excluded.bytes,
       packets = flow_buckets.packets + excluded.packets,
       out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
-      next_hop = CASE WHEN excluded.next_hop != '' THEN excluded.next_hop ELSE flow_buckets.next_hop END,
+      next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
       flow_start_ms = CASE
         WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
         THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
       flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
   `
-  lastFlushUsedTransaction = false
+
+async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise<number> {
+  if (rows.length === 0) return 0
   try {
-    for (const r of rows) {
-      await ensureParentPartition("flow_buckets", r.bucketAt)
-      await dbQuery(upsertSql, {
-        serverId: r.serverId,
-        bucketAt: r.bucketAt,
-        src: r.src,
-        dst: r.dst,
-        proto: r.proto,
-        srcPort: r.srcPort,
-        dstPort: r.dstPort,
-        bytes: r.bytes,
-        packets: r.packets,
-        inIface: r.inIface,
-        outIface: r.outIface,
-        nextHop: r.nextHop,
-        flowStartMs: r.flowStartMs,
-        flowEndMs: r.flowEndMs,
-      })
-    }
-    lastFlushUsedTransaction = true
-    rowsStored += rows.length
-    bumpDataEpoch()
-  } catch {
+    await upsertFlowBucketsBatch(rows)
+    return rows.length
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    setEngineError(`flow_buckets: ${message}`)
+    let stored = 0
     for (const r of rows) {
       try {
         await ensureParentPartition("flow_buckets", r.bucketAt)
-        await dbQuery(upsertSql, {
-          serverId: r.serverId,
-          bucketAt: r.bucketAt,
-          src: r.src,
-          dst: r.dst,
-          proto: r.proto,
-          srcPort: r.srcPort,
-          dstPort: r.dstPort,
-          bytes: r.bytes,
-          packets: r.packets,
-          inIface: r.inIface,
-          outIface: r.outIface,
-          nextHop: r.nextHop,
-          flowStartMs: r.flowStartMs,
-          flowEndMs: r.flowEndMs,
-        })
-        rowsStored += 1
-      } catch {
-        /* ignore single-row failures */
+        await dbQuery(FLOW_UPSERT_SQL, flowUpsertParams(r))
+        stored += 1
+      } catch (rowErr) {
+        if (stored === 0 && lastError.startsWith("flow_buckets:")) {
+          const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr)
+          setEngineError(`flow_buckets: ${rowMsg}`)
+        }
       }
     }
+    return stored
+  }
+}
+
+export async function flushPending(opts?: { force?: boolean }): Promise<void> {
+  pruneRecent()
+  rollFlowRings()
+  const force = Boolean(opts?.force)
+  const hasWork = pending.size > 0 || minuteRollup.size > 0 || minuteDims.size > 0
+  const due = persistDue(force, hasWork)
+  try {
+    await persistListenerStats(force)
+  } catch {
+    /* settings row may be absent in unit tests */
+  }
+
+  if (!hasWork) {
+    if (force) {
+      try {
+        await pruneStored()
+      } catch {
+        /* prune best-effort */
+      }
+    }
+    lastFlushUsedTransaction = false
+    return
+  }
+  if (!due) {
+    lastFlushUsedTransaction = false
+    return
+  }
+
+  const sanitized: PendingFlowRow[] = []
+  let skippedInet = 0
+  for (const row of topNPending([...pending.values()].map(toPendingRow))) {
+    const clean = sanitizeFlowRow(row)
+    if (!clean) {
+      skippedInet += 1
+      continue
+    }
+    sanitized.push(clean)
+  }
+  pending.clear()
+  for (const row of sanitized) mergeInto(recent, row)
+  if (skippedInet > 0) {
+    setEngineError(`flow_buckets: пропуск ${skippedInet} строк с невалидным IP`)
+  }
+
+  lastFlushUsedTransaction = false
+  lastPersistAt = Date.now()
+  lastFlushedMinute = minuteBucketIso()
+  try {
+    const stored = await upsertFlowBuckets(sanitized)
+    lastFlushUsedTransaction = stored === sanitized.length
+    rowsStored += stored
+    if (stored > 0) bumpDataEpoch()
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    setEngineError(`flow_buckets: ${message}`)
   }
   try {
     await upsertMinuteAndDaily()
@@ -730,7 +884,11 @@ export async function flushPending(): Promise<void> {
   } catch {
     /* rollup best-effort */
   }
-  await pruneStored()
+  try {
+    await pruneStored()
+  } catch {
+    /* prune best-effort */
+  }
 }
 
 export function lastFlushUsedTransactionForTests(): boolean {
@@ -738,7 +896,7 @@ export function lastFlushUsedTransactionForTests(): boolean {
 }
 
 export async function flushPendingForTests(): Promise<void> {
-  await flushPending()
+  await flushPending({ force: true })
 }
 
 export function onEngineTick(): void {
@@ -766,6 +924,9 @@ export function resetEngineForTests(): void {
   lastFlushUsedTransaction = false
   lastPruneAt = 0
   lastPassiveCheckpointAt = Date.now()
+  lastPersistAt = 0
+  lastFlushedMinute = ""
+  lastStatsPersistAt = 0
   lastPersistedStats = null
   bumpDataEpoch()
   pendingCap = MAX_PENDING

backend/src/services/traffic-flow-ingest.test.ts

@@ -21,7 +21,7 @@ import {
   setWantListenForTests,
   simulateWorkerExitForTests,
 } from "./traffic-flow-ingest.js"
-import { configureEngine, droppedForTests, pendingSizeForTests } from "./traffic-flow-engine.js"
+import { configureEngine, droppedForTests, getEngineStats, isValidFlowInet, pendingSizeForTests } from "./traffic-flow-engine.js"
 import { dbQuery } from "../db/index.js"
 import { withPgOrSkip } from "../test/pg.js"
 
@@ -69,6 +69,27 @@ assert.equal(droppedForTests(), 3)
 assert.equal(peekPendingFlows().length, 3)
 setPendingCapForTests(null)
 
+assert.equal(isValidFlowInet("10.0.0.1"), true)
+assert.equal(isValidFlowInet("8.8.8.8"), true)
+assert.equal(isValidFlowInet("0:0:0:0:0:0:0:1"), true)
+assert.equal(isValidFlowInet("not-an-ip"), false)
+assert.equal(isValidFlowInet("999.1.1.1"), false)
+
+resetFlowRingsForTests()
+ingestParsedFlowsForServerForTests(1, [{
+  src: "not-an-ip",
+  dst: "8.8.8.8",
+  proto: 6,
+  srcPort: 1,
+  dstPort: 443,
+  bytes: 10,
+  packets: 1,
+  inIface: "2",
+  outIface: "",
+}])
+await flushPendingForTests()
+assert.match(getEngineStats().lastError, /невалидн/)
+
 resetFlowRingsForTests()
 configureEngine({ topN: 20 })
 const talkers = Array.from({ length: 25 }, (_, i) => ({

backend/src/services/traffic-flow-ingest.ts

@@ -267,7 +267,7 @@ export async function startTrafficFlowListener() {
 export function stopTrafficFlowListener() {
   wantListen = false
   stopWorkerProcess()
-  void flushPending().catch(() => { /* ignore */ })
+  void flushPending({ force: true }).catch(() => { /* ignore */ })
   state = { bound: false, address: null }
 }
 
@@ -425,7 +425,7 @@ export async function ingestParsedFlowsForTests(exporterIp: string, flows: Parse
   if (serverId == null) return
   queueParsedFlows(serverId, flows)
   rollFlowRings()
-  await flushPending()
+  await flushPending({ force: true })
 }
 
 export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlowInput[]) {
@@ -444,7 +444,7 @@ export function lastFlushUsedTransactionForTests(): boolean {
 }
 
 export async function flushPendingForTests(): Promise<void> {
-  await flushPending()
+  await flushPending({ force: true })
 }
 
 async function tableCount(name: string): Promise<number> {

backend/src/services/uptime-collector.ts

@@ -1,4 +1,4 @@
-import { and, asc, eq, gte, inArray, lt, max, or } from "drizzle-orm"
+import { and, asc, eq, gte, inArray, max, or } from "drizzle-orm"
 import { db } from "../db/index.js"
 import {
   servers,
@@ -63,12 +63,6 @@ export async function getSettings() {
   return (await db.select().from(uptimeSettings).where(eq(uptimeSettings.id, 1)).limit(1))[0]
 }
 
-async function cleanup(retentionDays: number) {
-  const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString()
-  await db.delete(uptimeProbeSamples).where(lt(uptimeProbeSamples.sampledAt, cutoff))
-  await db.delete(uptimeResourceSamples).where(lt(uptimeResourceSamples.sampledAt, cutoff))
-}
-
 export function parsePing(results: Array<{ time?: string; status?: string; sent?: string; received?: string; "packet-loss"?: string; "avg-rtt"?: string }>) {
   const sum = [...results].reverse().find((r) =>
     r.sent != null || r.received != null || r["packet-loss"] != null || r["avg-rtt"] != null,
@@ -166,8 +160,6 @@ export async function collectResourceSamplesOnce(): Promise<ResourcesRunSnapshot
           freeHddSpace: freeHdd,
           totalHddSpace: totalHdd,
           uptimeSeconds,
-          boardName: String(resource["board-name"] ?? ""),
-          rosVersion: String(resource["version"] ?? ""),
         })
         const memUsedMb = totalMem > 0 ? Math.round((totalMem - freeMem) / (1024 * 1024)) : 0
         const memTotalMb = totalMem > 0 ? Math.round(totalMem / (1024 * 1024)) : 0
@@ -198,8 +190,6 @@ export async function collectResourceSamplesOnce(): Promise<ResourcesRunSnapshot
           freeHddSpace: 0,
           totalHddSpace: 0,
           uptimeSeconds: 0,
-          boardName: "",
-          rosVersion: "",
         })
         snapshot.servers.push({
           serverId: s.id,
@@ -211,7 +201,6 @@ export async function collectResourceSamplesOnce(): Promise<ResourcesRunSnapshot
       }
     }
 
-    await cleanup(Math.max(1, settings.retentionDays))
     await db.update(uptimeSettings).set({
       lastCollectedAt: now,
       lastDurationMs: Date.now() - started,
@@ -324,7 +313,6 @@ export async function collectPingProbesOnce(): Promise<PingRunSnapshot> {
       }
     }
 
-    await cleanup(Math.max(1, settings.retentionDays))
     await db.update(uptimeSettings).set({
       lastCollectedAt: now,
       lastDurationMs: Date.now() - started,
@@ -399,7 +387,6 @@ export async function collectPingForProbeIds(probeIds: string[]): Promise<{ poll
       polled += 1
     }
 
-    await cleanup(Math.max(1, settings.retentionDays))
     return { polled }
   } finally {
     collectingPing = false

deploy/docker-compose.cdn-mm.yml

@@ -144,9 +144,24 @@ services:
       - -c
       - maintenance_work_mem=128MB
       - -c
-      - wal_compression=on
+      - wal_compression=lz4
+      - -c
+      - default_toast_compression=lz4
+      - -c
+      - io_method=worker
+      - -c
+      - io_workers=3
+      - -c
+      - effective_io_concurrency=200
+      - -c
+      - max_wal_size=2GB
+      - -c
+      - checkpoint_timeout=15min
+      - -c
+      - checkpoint_completion_target=0.9
     volumes:
-      - mmapp-pgdata:/var/lib/postgresql/data
+      # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker)
+      - mmapp-pgdata:/var/lib/postgresql
     networks:
       - mmapp
     healthcheck:

deploy/docker-compose.postgres.yml

@@ -30,9 +30,24 @@ services:
       - -c
       - maintenance_work_mem=128MB
       - -c
-      - wal_compression=on
+      - wal_compression=lz4
+      - -c
+      - default_toast_compression=lz4
+      - -c
+      - io_method=worker
+      - -c
+      - io_workers=3
+      - -c
+      - effective_io_concurrency=200
+      - -c
+      - max_wal_size=2GB
+      - -c
+      - checkpoint_timeout=15min
+      - -c
+      - checkpoint_completion_target=0.9
     volumes:
-      - mmapp-pgdata:/var/lib/postgresql/data
+      # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker)
+      - mmapp-pgdata:/var/lib/postgresql
     ports:
       - "127.0.0.1:${POSTGRES_PORT:-5432}:5432"
     healthcheck:

deploy/docker-compose.traefik-cdn.yml

@@ -46,9 +46,24 @@ services:
       - -c
       - maintenance_work_mem=128MB
       - -c
-      - wal_compression=on
+      - wal_compression=lz4
+      - -c
+      - default_toast_compression=lz4
+      - -c
+      - io_method=worker
+      - -c
+      - io_workers=3
+      - -c
+      - effective_io_concurrency=200
+      - -c
+      - max_wal_size=2GB
+      - -c
+      - checkpoint_timeout=15min
+      - -c
+      - checkpoint_completion_target=0.9
     volumes:
-      - mmapp-pgdata:/var/lib/postgresql/data
+      # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker)
+      - mmapp-pgdata:/var/lib/postgresql
     networks:
       - mmapp
     healthcheck:

deploy/docker-compose.traefik.yml

@@ -81,9 +81,24 @@ services:
       - -c
       - maintenance_work_mem=128MB
       - -c
-      - wal_compression=on
+      - wal_compression=lz4
+      - -c
+      - default_toast_compression=lz4
+      - -c
+      - io_method=worker
+      - -c
+      - io_workers=3
+      - -c
+      - effective_io_concurrency=200
+      - -c
+      - max_wal_size=2GB
+      - -c
+      - checkpoint_timeout=15min
+      - -c
+      - checkpoint_completion_target=0.9
     volumes:
-      - mmapp-pgdata:/var/lib/postgresql/data
+      # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker)
+      - mmapp-pgdata:/var/lib/postgresql
     networks:
       - mmapp
     healthcheck:

deploy/docker-compose.yml

@@ -26,9 +26,24 @@ services:
       - -c
       - maintenance_work_mem=128MB
       - -c
-      - wal_compression=on
+      - wal_compression=lz4
+      - -c
+      - default_toast_compression=lz4
+      - -c
+      - io_method=worker
+      - -c
+      - io_workers=3
+      - -c
+      - effective_io_concurrency=200
+      - -c
+      - max_wal_size=2GB
+      - -c
+      - checkpoint_timeout=15min
+      - -c
+      - checkpoint_completion_target=0.9
     volumes:
-      - mmapp-pgdata:/var/lib/postgresql/data
+      # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker)
+      - mmapp-pgdata:/var/lib/postgresql
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U mmapp -d mmapp"]
       interval: 5s

deploy/postgres.conf

@@ -0,0 +1,23 @@
+# MikrotikManager — PostgreSQL 18 runtime (overrides via `postgres -c`).
+# Не подменять целиком config_file образа: эти GUC передаются как -c в compose.
+
+timezone = UTC
+listen_addresses = '*'
+max_connections = 100
+shared_buffers = 256MB
+work_mem = 16MB
+maintenance_work_mem = 128MB
+
+# PG18 AIO: worker (io_uring в Docker/Alpine часто недоступен)
+io_method = worker
+io_workers = 3
+effective_io_concurrency = 200
+
+# TOAST: lz4 (не zstd — для колонок только pglz/lz4)
+default_toast_compression = lz4
+wal_compression = lz4
+
+# Реже checkpoint / меньше full-page writes (NetFlow upsert)
+max_wal_size = 2GB
+checkpoint_timeout = 15min
+checkpoint_completion_target = 0.9

deploy/run-beside-cdn-traefik.sh

@@ -97,14 +97,16 @@ if ! docker inspect mmapp-postgres >/dev/null 2>&1; then
     -e POSTGRES_INITDB_ARGS="--encoding=UTF8 --locale=C.UTF-8 --data-checksums" \
     -e TZ=UTC \
     -e PGTZ=UTC \
-    -v mmapp-pgdata:/var/lib/postgresql/data \
+    -v mmapp-pgdata:/var/lib/postgresql \
     --health-cmd="pg_isready -U mmapp -d mmapp" \
     --health-interval=5s \
     --health-timeout=5s \
     --health-retries=10 \
     postgres:18-alpine \
     postgres -c timezone=UTC -c listen_addresses=* -c max_connections=100 \
-      -c shared_buffers=256MB -c work_mem=16MB -c maintenance_work_mem=128MB -c wal_compression=on
+      -c shared_buffers=256MB -c work_mem=16MB -c maintenance_work_mem=128MB \
+      -c wal_compression=lz4 -c default_toast_compression=lz4 \
+      -c io_method=worker -c io_workers=3 -c effective_io_concurrency=200
 fi
 until docker exec mmapp-postgres pg_isready -U mmapp -d mmapp >/dev/null 2>&1; do
   sleep 1

hooks/use-flow-live.ts

@@ -15,6 +15,12 @@ function parseSseBlock(block: string): { event: string; data: string } {
   return { event, data: dataLines.join("\n") }
 }
 
+export function isValidFlowLiveSample(value: unknown): value is FlowAnalyticsDto {
+  if (!value || typeof value !== "object") return false
+  const v = value as Partial<FlowAnalyticsDto>
+  return typeof v.uniqueSrc === "number" && Array.isArray(v.destinations)
+}
+
 export function useFlowLive(opts: {
   enabled: boolean
   backendUrl: string
@@ -75,7 +81,11 @@ export function useFlowLive(opts: {
             if (!raw.trim() || raw.trim().startsWith(":")) continue
             const ev = parseSseBlock(raw)
             if (ev.event === "sample" && ev.data) {
-              const parsed = JSON.parse(ev.data) as FlowAnalyticsDto
+              const parsed = JSON.parse(ev.data) as unknown
+              if (!isValidFlowLiveSample(parsed)) {
+                setError("live sample пустой")
+                continue
+              }
               setSample(parsed)
               setError(parsed.degraded ? "Коллектор перегружен: упрощённая аналитика" : null)
             } else if (ev.event === "error" && ev.data) {