From c6c859a495438985cd53f8804403f2a9226bf4b5 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 8 Sep 2026 10:54:36 +0700 Subject: [PATCH] =?UTF-8?q?perf(db):=20=D1=81=D0=B6=D0=B0=D1=82=D1=8C=20?= =?UTF-8?q?=D1=81=D1=85=D0=B5=D0=BC=D1=83=20PostgreSQL=2018=20=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D0=BD=D0=BE=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=B3=D1=80=D1=83=D0=B7=D0=B8=D1=82=D1=8C=20SQLite?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit При первом рестарте wipe всех таблиц и импорт из SQLite в компактную схему. Retention через DROP PARTITION, lz4 и AIO worker. Co-authored-by: Cursor --- backend/drizzle/0002_compact_schema.sql | 620 ++++++++++++++++++ backend/package.json | 2 +- backend/src/db/bootstrap.ts | 11 +- backend/src/db/migrate.ts | 35 +- backend/src/db/partitions.ts | 56 +- backend/src/db/pg-schema.test.ts | 90 +++ backend/src/db/schema.ts | 31 +- backend/src/db/sqlite-import.ts | 107 ++- backend/src/db/traffic-flags.test.ts | 19 + backend/src/db/traffic-flags.ts | 24 + .../modules/users/service/users-service.ts | 9 +- backend/src/routes/ospf.ts | 2 +- backend/src/routes/uptime.ts | 13 +- .../src/services/internet-path-collector.ts | 8 +- backend/src/services/poller.ts | 17 +- .../services/servers-rest-ping-collector.ts | 2 - backend/src/services/traffic-collector.ts | 19 +- backend/src/services/traffic-flow-engine.ts | 60 +- backend/src/services/uptime-collector.ts | 15 +- deploy/docker-compose.cdn-mm.yml | 10 +- deploy/docker-compose.postgres.yml | 10 +- deploy/docker-compose.traefik-cdn.yml | 10 +- deploy/docker-compose.traefik.yml | 10 +- deploy/docker-compose.yml | 10 +- deploy/postgres.conf | 18 + deploy/run-beside-cdn-traefik.sh | 4 +- 26 files changed, 1055 insertions(+), 157 deletions(-) create mode 100644 backend/drizzle/0002_compact_schema.sql create mode 100644 backend/src/db/traffic-flags.test.ts create mode 100644 backend/src/db/traffic-flags.ts create mode 100644 deploy/postgres.conf diff --git a/backend/drizzle/0002_compact_schema.sql b/backend/drizzle/0002_compact_schema.sql new file mode 100644 index 0000000..9e9de78 --- /dev/null +++ b/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; diff --git a/backend/package.json b/backend/package.json index 66e6340..43b42c1 100644 --- a/backend/package.json +++ b/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/sqlite-json.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": { diff --git a/backend/src/db/bootstrap.ts b/backend/src/db/bootstrap.ts index ef21c3a..2ab8feb 100644 --- a/backend/src/db/bootstrap.ts +++ b/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 { 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 { diff --git a/backend/src/db/migrate.ts b/backend/src/db/migrate.ts index de3e165..bf858b8 100644 --- a/backend/src/db/migrate.ts +++ b/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 { await pool.query(` CREATE TABLE IF NOT EXISTS schema_migrations ( @@ -30,14 +34,21 @@ export async function applySqlMigrations(pool: Pool): Promise { 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], + ) + } } diff --git a/backend/src/db/partitions.ts b/backend/src/db/partitions.ts index 0fe7490..4f85160 100644 --- a/backend/src/db/partitions.ts +++ b/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> { + 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 { + 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 { + 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 { + 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 diff --git a/backend/src/db/pg-schema.test.ts b/backend/src/db/pg-schema.test.ts index 74a559f..694143c 100644 --- a/backend/src/db/pg-schema.test.ts +++ b/backend/src/db/pg-schema.test.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict" import { withPgOrSkip } from "../test/pg.js" import { dbQuery, pool } from "./index.js" +import { applySqlMigrations } from "./migrate.js" import { ensurePartitionFor } from "./partitions.js" if (!(await withPgOrSkip())) { @@ -80,4 +81,93 @@ if (!(await withPgOrSkip())) { assert.equal(arrayAsPgArrayFailed, true, "JS array must not be bound as jsonb without stringify") } +{ + const { rows } = await dbQuery<{ attname: string }>(` + SELECT a.attname + FROM pg_index i + JOIN unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord) ON true + JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum + WHERE i.indrelid = 'traffic_samples'::regclass AND i.indisprimary + ORDER BY k.ord + `) + assert.deepEqual( + rows.map((r) => r.attname), + ["server_id", "sampled_at", "interface_name", "peer_public_key"], + "traffic_samples PK без id", + ) +} + +{ + const { rows } = await dbQuery<{ column_name: string }>(` + SELECT column_name FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'traffic_samples' + `) + const cols = new Set(rows.map((r) => r.column_name)) + assert.equal(cols.has("id"), false) + assert.equal(cols.has("running"), false) + assert.equal(cols.has("disabled"), false) + assert.equal(cols.has("flags"), true) +} + +{ + const { rows } = await dbQuery<{ column_name: string; udt_name: string }>(` + SELECT column_name, udt_name + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'flow_buckets' + AND column_name IN ('src', 'dst', 'next_hop', 'proto') + `) + const by = Object.fromEntries(rows.map((r) => [r.column_name, r.udt_name])) + assert.equal(by.src, "inet") + assert.equal(by.dst, "inet") + assert.equal(by.next_hop, "inet") + assert.equal(by.proto, "int2") +} + +{ + const { rows } = await dbQuery<{ indexdef: string }>(` + SELECT indexdef FROM pg_indexes + WHERE schemaname = 'public' AND tablename = 'traffic_samples' + `) + const defs = rows.map((r) => r.indexdef.toLowerCase()) + assert.equal(defs.some((d) => d.includes("using brin")), false, "нет BRIN на traffic_samples") + assert.equal( + defs.filter((d) => d.includes("idx_traffic_samples_server_iface_time")).length, + 1, + "один btree (server_id, interface_name, sampled_at)", + ) + assert.equal(defs.some((d) => d.includes("idx_traffic_samples_server_time")), false) +} + +{ + const { rows } = await dbQuery<{ column_name: string }>(` + SELECT column_name FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'uptime_resource_samples' + `) + const cols = new Set(rows.map((r) => r.column_name)) + assert.equal(cols.has("board_name"), false) + assert.equal(cols.has("ros_version"), false) +} + +{ + const mig = await dbQuery<{ id: string }>( + `SELECT id FROM schema_migrations WHERE id = '0002_compact_schema'`, + ) + assert.equal(mig.rows.length, 1, "0002 применена") + + await dbQuery(`INSERT INTO servers (name, host) VALUES ('pg-wipe-idempotent', '127.0.0.1')`) + await applySqlMigrations(pool) + const still = await dbQuery<{ n: string }>( + `SELECT COUNT(*)::text AS n FROM servers WHERE name = 'pg-wipe-idempotent'`, + ) + assert.equal(still.rows[0]?.n, "1", "повторный applySqlMigrations не wipe") + await dbQuery(`DELETE FROM servers WHERE name = 'pg-wipe-idempotent'`) +} + +{ + const marker = await dbQuery<{ sqlite_imported_at: string | null }>( + `SELECT sqlite_imported_at FROM data_migration WHERE id = 1`, + ) + assert.ok(marker.rows[0], "data_migration singleton после wipe") +} + console.log("pg-schema.test.ts: ok") diff --git a/backend/src/db/schema.ts b/backend/src/db/schema.ts index fc3c208..5aaf429 100644 --- a/backend/src/db/schema.ts +++ b/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", { diff --git a/backend/src/db/sqlite-import.ts b/backend/src/db/sqlite-import.ts index 18b174d..78119c4 100644 --- a/backend/src/db/sqlite-import.ts +++ b/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, +): unknown { switch (kind) { case "ts": return parseTs(value, strict, rejects, ctx) @@ -303,6 +319,10 @@ function coerce(kind: ColKind, value: unknown, strict: boolean, rejects: string[ return Number(value) case "text": return value == null ? "" : String(value) + case "flags": + return encodeTrafficFlags(parseBool(row?.running), parseBool(row?.disabled)) + case "inet": + return parseInet(value) default: return value == null ? null : String(value) } @@ -326,6 +346,35 @@ async function setval(pool: Pool, table: string): Promise { ) } +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 { + const part = specForParent(spec.table) + if (!part || !spec.timeCol) return + const bounds = sqlite.prepare( + `SELECT MIN(${spec.timeCol}) AS a, MAX(${spec.timeCol}) AS b FROM ${spec.table}${where}`, + ).get() as { a?: unknown; b?: unknown } + if (bounds?.a == null || bounds?.b == null) return + const isDate = spec.columns.find((c) => c[0] === spec.timeCol)?.[1] === "date" + const fromIso = isDate + ? `${String(bounds.a).slice(0, 10)}T00:00:00Z` + : parseTs(bounds.a, false, [], spec.table) + const toIso = isDate + ? `${String(bounds.b).slice(0, 10)}T00:00:00Z` + : parseTs(bounds.b, false, [], spec.table) + if (!fromIso || !toIso) return + await ensurePartitionsBetween(pool, spec.table, part.kind, new Date(fromIso), new Date(toIso)) +} + async function copyTable( sqlite: Database.Database, pool: Pool, @@ -341,28 +390,27 @@ async function copyTable( where = ` WHERE ${spec.timeCol} >= '${cutoff.replace("T", " ").slice(0, 19)}' OR ${spec.timeCol} >= '${cutoff}'` } const total = (sqlite.prepare(`SELECT COUNT(*) AS c FROM ${spec.table}${where}`).get() as { c: number }).c - const part = specForParent(spec.table) + await precreatePartitions(sqlite, pool, spec, where) const cols = spec.columns.map(([c]) => c) - const placeholders = spec.columns - .map(([, kind], i) => (kind === "json" || kind === "json-null" ? `$${i + 1}::jsonb` : `$${i + 1}`)) - .join(", ") const conflictSql = spec.upsert ? `ON CONFLICT (id) DO UPDATE SET ${cols.filter((c) => c !== "id").map((c) => `${c} = EXCLUDED.${c}`).join(", ")}` : `ON CONFLICT DO NOTHING` - const insertSql = `INSERT INTO ${spec.table} (${cols.join(", ")}) VALUES (${placeholders}) ${conflictSql}` let copied = 0 let skipped = 0 const stmt = sqlite.prepare(`SELECT * FROM ${spec.table}${where}`) const batch: unknown[][] = [] const flush = async () => { if (batch.length === 0) return + const n = spec.columns.length + const valuesSql = batch.map((_, i) => + `(${spec.columns.map(([, kind], j) => placeholderFor(kind, i * n + j + 1)).join(", ")})`, + ).join(", ") + const insertSql = `INSERT INTO ${spec.table} (${cols.join(", ")}) VALUES ${valuesSql} ${conflictSql}` const client = await pool.connect() try { await client.query("BEGIN") - for (const values of batch) { - await client.query(insertSql, values) - copied += 1 - } + await client.query(insertSql, batch.flat()) + copied += batch.length await client.query("COMMIT") } catch (err) { await client.query("ROLLBACK") @@ -375,22 +423,15 @@ async function copyTable( } for (const row of stmt.iterate() as Iterable>) { 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)}` diff --git a/backend/src/db/traffic-flags.test.ts b/backend/src/db/traffic-flags.test.ts new file mode 100644 index 0000000..1fd05e4 --- /dev/null +++ b/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") diff --git a/backend/src/db/traffic-flags.ts b/backend/src/db/traffic-flags.ts new file mode 100644 index 0000000..a0a813b --- /dev/null +++ b/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), + } +} diff --git a/backend/src/modules/users/service/users-service.ts b/backend/src/modules/users/service/users-service.ts index 999adcf..cb25f81 100644 --- a/backend/src/modules/users/service/users-service.ts +++ b/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 { return latest.length > 0 ? Math.max(1, Math.round(latest[0].latencyMs ?? 100)) : 100 } -async function latestTrafficByInterface(serverId: number): Promise> { +async function latestTrafficByInterface(serverId: number): Promise> { const rows = await db .select() .from(trafficSamples) diff --git a/backend/src/routes/uptime.ts b/backend/src/routes/uptime.ts index 30f1cea..cdfb9a8 100644 --- a/backend/src/routes/uptime.ts +++ b/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, } })) diff --git a/backend/src/services/internet-path-collector.ts b/backend/src/services/internet-path-collector.ts index 2ab568e..4a44e9f 100644 --- a/backend/src/services/internet-path-collector.ts +++ b/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 { .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 ───────────────────────────────────────────────────────────────────── diff --git a/backend/src/services/servers-rest-ping-collector.ts b/backend/src/services/servers-rest-ping-collector.ts index 4a066e7..bd7135b 100644 --- a/backend/src/services/servers-rest-ping-collector.ts +++ b/backend/src/services/servers-rest-ping-collector.ts @@ -162,8 +162,6 @@ export async function collectServersRestPingOnce(): Promise> { const last = (await db .select({ sampledAt: trafficSamples.sampledAt }) @@ -164,8 +159,7 @@ export async function collectTrafficOnce(): Promise { txBytes, rxBps, txBps, - running, - disabled, + flags: encodeTrafficFlags(running, disabled), } }) try { @@ -194,8 +188,7 @@ export async function collectTrafficOnce(): Promise { txBytes, rxBps, txBps, - running, - disabled, + flags: encodeTrafficFlags(running, disabled), }) } } catch { @@ -224,7 +217,6 @@ export async function collectTrafficOnce(): Promise { } } - 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) })) } diff --git a/backend/src/services/traffic-flow-engine.ts b/backend/src/services/traffic-flow-engine.ts index 4e76f68..567e54b 100644 --- a/backend/src/services/traffic-flow-engine.ts +++ b/backend/src/services/traffic-flow-engine.ts @@ -48,6 +48,30 @@ export interface PendingFlowRow { flowEndMs: number } +function inetOrNull(value: string | null | undefined): string | null { + const s = String(value ?? "").trim() + return s.length > 0 ? s : null +} + +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 @@ -668,7 +692,7 @@ export async function flushPending(): Promise { 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, @@ -678,22 +702,7 @@ export async function flushPending(): Promise { 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, - }) + await dbQuery(upsertSql, flowUpsertParams(r)) } lastFlushUsedTransaction = true rowsStored += rows.length @@ -702,22 +711,7 @@ export async function flushPending(): Promise { 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, - }) + await dbQuery(upsertSql, flowUpsertParams(r)) rowsStored += 1 } catch { /* ignore single-row failures */ diff --git a/backend/src/services/uptime-collector.ts b/backend/src/services/uptime-collector.ts index f37a92a..dd51149 100644 --- a/backend/src/services/uptime-collector.ts +++ b/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 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 { } } - 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 diff --git a/deploy/docker-compose.cdn-mm.yml b/deploy/docker-compose.cdn-mm.yml index b3a98d0..bef53c4 100644 --- a/deploy/docker-compose.cdn-mm.yml +++ b/deploy/docker-compose.cdn-mm.yml @@ -144,7 +144,15 @@ 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 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.postgres.yml b/deploy/docker-compose.postgres.yml index 671ae77..8045a9f 100644 --- a/deploy/docker-compose.postgres.yml +++ b/deploy/docker-compose.postgres.yml @@ -30,7 +30,15 @@ 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 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.traefik-cdn.yml b/deploy/docker-compose.traefik-cdn.yml index 1485bb4..1718cf2 100644 --- a/deploy/docker-compose.traefik-cdn.yml +++ b/deploy/docker-compose.traefik-cdn.yml @@ -46,7 +46,15 @@ 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 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.traefik.yml b/deploy/docker-compose.traefik.yml index 3cce3ed..61064dd 100644 --- a/deploy/docker-compose.traefik.yml +++ b/deploy/docker-compose.traefik.yml @@ -81,7 +81,15 @@ 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 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 45ba7ae..5043890 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -26,7 +26,15 @@ 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 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/postgres.conf b/deploy/postgres.conf new file mode 100644 index 0000000..0a810cb --- /dev/null +++ b/deploy/postgres.conf @@ -0,0 +1,18 @@ +# 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 diff --git a/deploy/run-beside-cdn-traefik.sh b/deploy/run-beside-cdn-traefik.sh index 91a4996..ddfac01 100644 --- a/deploy/run-beside-cdn-traefik.sh +++ b/deploy/run-beside-cdn-traefik.sh @@ -104,7 +104,9 @@ if ! docker inspect mmapp-postgres >/dev/null 2>&1; then --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