From c077c209e5809d4bdde3cf9a836ea53d9801c137 Mon Sep 17 00:00:00 2001 From: Laan Tungir Date: Wed, 1 Apr 2026 05:25:03 -0400 Subject: [PATCH] v2.0.0 - Major architecture refactor: db_ops abstraction, low-risk SQLite refactor, and thread-pool scaffolding --- Makefile | 2 +- plans/c_relay_pg_plan.md | 632 ++++++++++++++ plans/c_relay_thread_pool_plan.md | 332 +++++++ plans/database_architecture_analysis.md | 1055 +++++++++++++++++++++++ src/db_ops.c | 292 +++++++ src/db_ops.h | 31 + src/ip_ban.c | 13 +- src/ip_ban.h | 7 +- src/main.c | 49 +- src/main.h | 8 +- src/nip009.c | 122 +-- src/request_validator.c | 112 +-- src/subscriptions.c | 149 +--- src/thread_pool.c | 276 ++++++ src/thread_pool.h | 70 ++ src/websockets.c | 11 +- 16 files changed, 2832 insertions(+), 329 deletions(-) create mode 100644 plans/c_relay_pg_plan.md create mode 100644 plans/c_relay_thread_pool_plan.md create mode 100644 plans/database_architecture_analysis.md create mode 100644 src/db_ops.c create mode 100644 src/db_ops.h create mode 100644 src/thread_pool.c create mode 100644 src/thread_pool.h diff --git a/Makefile b/Makefile index 8132f46..509a977 100644 --- a/Makefile +++ b/Makefile @@ -9,7 +9,7 @@ LIBS = -lsqlite3 -lwebsockets -lz -ldl -lpthread -lm -L/usr/local/lib -lsecp256k BUILD_DIR = build # Source files -MAIN_SRC = src/main.c src/config.c src/dm_admin.c src/request_validator.c src/nip009.c src/nip011.c src/nip013.c src/nip040.c src/nip042.c src/websockets.c src/subscriptions.c src/api.c src/embedded_web_content.c src/ip_ban.c +MAIN_SRC = src/main.c src/config.c src/dm_admin.c src/request_validator.c src/nip009.c src/nip011.c src/nip013.c src/nip040.c src/nip042.c src/websockets.c src/subscriptions.c src/api.c src/embedded_web_content.c src/ip_ban.c src/db_ops.c src/thread_pool.c NOSTR_CORE_LIB = nostr_core_lib/libnostr_core_x64.a C_UTILS_LIB = c_utils_lib/libc_utils.a diff --git a/plans/c_relay_pg_plan.md b/plans/c_relay_pg_plan.md new file mode 100644 index 0000000..53e3d1b --- /dev/null +++ b/plans/c_relay_pg_plan.md @@ -0,0 +1,632 @@ +# c-relay-pg Plan — PostgreSQL Backend + Multi-Instance + Dashboard + +## Overview + +**c-relay-pg** is a new project (separate repository) forked from c-relay after the `db_ops` abstraction layer is complete. It replaces the SQLite backend with PostgreSQL, enabling horizontal scaling, external dashboards, and production-grade deployments. + +### Prerequisites + +- The `db_ops.h` / `db_ops.c` abstraction layer must be complete in c-relay first — see [c-relay thread pool plan](c_relay_thread_pool_plan.md) Phase 1. +- The fork happens after Phase 1 of that plan is done and tested. + +### Why a Separate Project? + +| | c-relay | c-relay-pg | +|---|---------|-----------| +| **Database** | SQLite (embedded) | PostgreSQL (client-server) | +| **Deployment** | Single binary + DB file | Binary + PostgreSQL server | +| **Scaling** | Single instance | Multiple instances + load balancer | +| **Dashboard** | Embedded web UI | Separate web server + Grafana | +| **Target user** | Personal relay, small community | Medium-large relay, production | +| **Complexity** | Minimal | More infrastructure | + +For the full analysis of why PostgreSQL was chosen over MySQL/MariaDB and other databases, see the [database architecture analysis](database_architecture_analysis.md). + +## Architecture + +```mermaid +flowchart TD + subgraph c-relay-pg - Production Deployment + LB[nginx - TLS + load balancing] -->|WebSocket| R1[c-relay-pg Instance 1] + LB -->|WebSocket| R2[c-relay-pg Instance 2] + LB -->|HTTP| DASH[Dashboard] + R1 -->|db_ops API| PGBACK[PostgreSQL Backend - db_ops_postgres.c] + R2 -->|db_ops API| PGBACK + PGBACK -->|libpq| PG[PostgreSQL Server] + DASH -->|SQL| PG + R1 <-->|LISTEN/NOTIFY| R2 + end +``` + +--- + +## Phase 1: PostgreSQL Backend + +### Goal + +Fork c-relay into a new repository called **c-relay-pg**. Replace the SQLite `db_ops` implementation with PostgreSQL using `libpq`. The `db_ops.h` interface stays identical — only the backend changes. + +### New Files + +- `src/db_ops_sqlite.c` — Renamed from `db_ops.c` (the Phase 1 SQLite implementation from c-relay) +- `src/db_ops_postgres.c` — New PostgreSQL implementation +- `src/db_ops.c` — Thin dispatcher that calls the active backend + +### PostgreSQL Schema + +```sql +-- PostgreSQL schema for c-relay-pg +-- No event_tags table needed — JSONB + GIN handles everything + +CREATE TABLE events ( + id TEXT PRIMARY KEY, + pubkey TEXT NOT NULL, + created_at BIGINT NOT NULL, + kind INTEGER NOT NULL, + event_type TEXT NOT NULL CHECK (event_type IN ('regular', 'replaceable', 'ephemeral', 'addressable')), + content TEXT NOT NULL, + sig TEXT NOT NULL, + tags JSONB NOT NULL DEFAULT '[]', + event_json TEXT NOT NULL, + first_seen BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT +); + +-- Core indexes +CREATE INDEX idx_events_pubkey ON events(pubkey); +CREATE INDEX idx_events_kind ON events(kind); +CREATE INDEX idx_events_created_at ON events(created_at DESC); +CREATE INDEX idx_events_kind_created_at ON events(kind, created_at DESC); +CREATE INDEX idx_events_pubkey_created_at ON events(pubkey, created_at DESC); +CREATE INDEX idx_events_pubkey_kind ON events(pubkey, kind); + +-- THE KEY INDEX: GIN on JSONB tags — replaces the entire event_tags table +CREATE INDEX idx_events_tags ON events USING GIN (tags); + +-- Partial index: only non-ephemeral events (what REQ queries actually filter) +CREATE INDEX idx_events_non_ephemeral ON events(created_at DESC) + WHERE kind < 20000 OR kind >= 30000; + +-- Config table +CREATE TABLE config ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + data_type TEXT NOT NULL CHECK (data_type IN ('string', 'integer', 'boolean', 'json')), + description TEXT, + category TEXT DEFAULT 'general', + requires_restart INTEGER DEFAULT 0, + created_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT, + updated_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT +); + +-- Auth rules table +CREATE TABLE auth_rules ( + id SERIAL PRIMARY KEY, + rule_type TEXT NOT NULL CHECK (rule_type IN ('whitelist', 'blacklist', 'rate_limit', 'auth_required', 'wot_whitelist')), + pattern_type TEXT NOT NULL CHECK (pattern_type IN ('pubkey', 'kind', 'ip', 'global')), + pattern_value TEXT, + active INTEGER NOT NULL DEFAULT 1, + created_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT, + updated_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT +); + +CREATE INDEX idx_auth_rules_lookup ON auth_rules(rule_type, pattern_type, pattern_value) WHERE active = 1; + +-- Relay private key storage +CREATE TABLE relay_seckey ( + private_key_hex TEXT NOT NULL CHECK (length(private_key_hex) = 64), + created_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT +); + +-- Subscription logging +CREATE TABLE subscriptions ( + id SERIAL PRIMARY KEY, + subscription_id TEXT NOT NULL, + wsi_pointer TEXT NOT NULL, + client_ip TEXT NOT NULL, + event_type TEXT NOT NULL CHECK (event_type IN ('created', 'closed', 'expired', 'disconnected')), + filter_json TEXT, + events_sent INTEGER DEFAULT 0, + created_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT, + ended_at BIGINT, + duration INTEGER, + UNIQUE(subscription_id, wsi_pointer) +); + +-- IP ban persistence +CREATE TABLE ip_bans ( + ip TEXT PRIMARY KEY, + failure_count INTEGER NOT NULL DEFAULT 0, + ban_count INTEGER NOT NULL DEFAULT 0, + banned_until BIGINT NOT NULL DEFAULT 0, + first_failure BIGINT NOT NULL DEFAULT 0, + has_authed_successfully INTEGER NOT NULL DEFAULT 0, + last_success_at BIGINT NOT NULL DEFAULT 0, + total_connections INTEGER NOT NULL DEFAULT 0, + total_failures INTEGER NOT NULL DEFAULT 0, + total_successes INTEGER NOT NULL DEFAULT 0, + first_seen BIGINT NOT NULL DEFAULT 0, + idle_failure_count INTEGER NOT NULL DEFAULT 0, + idle_ban_count INTEGER NOT NULL DEFAULT 0, + idle_banned_until BIGINT NOT NULL DEFAULT 0, + idle_first_failure BIGINT NOT NULL DEFAULT 0 +); + +-- Materialized views for dashboard (refreshed periodically) +CREATE MATERIALIZED VIEW event_kinds_mv AS +SELECT kind, COUNT(*) as count, + ROUND(COUNT(*) * 100.0 / NULLIF((SELECT COUNT(*) FROM events), 0), 2) as percentage +FROM events GROUP BY kind; + +CREATE MATERIALIZED VIEW time_stats_mv AS +SELECT 'total' as period, COUNT(*) as total_events, COUNT(DISTINCT pubkey) as unique_pubkeys +FROM events +UNION ALL +SELECT '24h', COUNT(*), COUNT(DISTINCT pubkey) +FROM events WHERE created_at >= EXTRACT(EPOCH FROM NOW())::BIGINT - 86400 +UNION ALL +SELECT '7d', COUNT(*), COUNT(DISTINCT pubkey) +FROM events WHERE created_at >= EXTRACT(EPOCH FROM NOW())::BIGINT - 604800; + +CREATE MATERIALIZED VIEW top_pubkeys_mv AS +SELECT pubkey, COUNT(*) as event_count, + ROUND(COUNT(*) * 100.0 / NULLIF((SELECT COUNT(*) FROM events), 0), 2) as percentage +FROM events GROUP BY pubkey ORDER BY event_count DESC LIMIT 20; + +-- Unique indexes required for CONCURRENTLY refresh +CREATE UNIQUE INDEX idx_event_kinds_mv ON event_kinds_mv(kind); +CREATE UNIQUE INDEX idx_time_stats_mv ON time_stats_mv(period); +CREATE UNIQUE INDEX idx_top_pubkeys_mv ON top_pubkeys_mv(pubkey); +``` + +### Key Implementation Details + +#### Tag Queries with JSONB + +The biggest win — replacing the `event_tags` subquery with JSONB containment: + +```c +// SQLite (current): subquery into event_tags table +// AND id IN (SELECT event_id FROM event_tags WHERE tag_name = ? AND tag_value IN (?)) + +// PostgreSQL: JSONB containment operator with GIN index +// AND tags @> '[["p", "pubkey_hex"]]' + +// For multiple tag values: +// AND (tags @> '[["p", "val1"]]' OR tags @> '[["p", "val2"]]') +``` + +This eliminates the entire `event_tags` table (4 million rows, ~2 GB indexes in the current SQLite schema). The GIN index on JSONB handles everything in ~100–200 MB. + +#### LISTEN/NOTIFY Integration + +For cross-instance event broadcasting: + +```c +// In db_ops_postgres.c: +int db_notify_new_event(const char* event_id) { + char cmd[128]; + snprintf(cmd, sizeof(cmd), "NOTIFY new_event, '%s'", event_id); + PGresult* res = PQexec(g_pg_conn, cmd); + PQclear(res); + return 0; +} + +// Called from the lws event loop every iteration: +int db_check_notifications(char* event_id_out, size_t max_len) { + PQconsumeInput(g_pg_notify_conn); + PGnotify* notify = PQnotifies(g_pg_notify_conn); + if (notify) { + strncpy(event_id_out, notify->extra, max_len); + PQfreemem(notify); + return 1; // Got a notification + } + return 0; // No notifications +} +``` + +#### Connection Management + +```c +// db_ops_postgres.c connection pool (simple version) +#define DB_POOL_SIZE 4 + +static PGconn* g_pg_read_pool[DB_POOL_SIZE]; // For REQ queries +static PGconn* g_pg_write_conn; // For EVENT inserts +static PGconn* g_pg_notify_conn; // For LISTEN/NOTIFY +static pthread_mutex_t g_pool_lock; + +PGconn* db_get_read_connection(void) { + pthread_mutex_lock(&g_pool_lock); + // Round-robin or find idle connection + // ... + pthread_mutex_unlock(&g_pool_lock); +} +``` + +### Build System Changes + +```makefile +# Makefile additions +DB_BACKEND ?= sqlite # Default to sqlite, override with: make DB_BACKEND=postgres + +ifeq ($(DB_BACKEND),postgres) + MAIN_SRC += src/db_ops.c src/db_ops_postgres.c + LIBS += -lpq + CFLAGS += -DDB_BACKEND_POSTGRES +else + MAIN_SRC += src/db_ops.c src/db_ops_sqlite.c + CFLAGS += -DDB_BACKEND_SQLITE +endif +``` + +### Data Migration Tool + +A standalone tool to migrate existing SQLite data to PostgreSQL: + +```bash +# migrate_to_postgres.sh +# 1. Export events from SQLite +sqlite3 old_relay.db ".mode csv" "SELECT id,pubkey,created_at,kind,event_type,content,sig,tags,event_json,first_seen FROM events" > events.csv + +# 2. Import into PostgreSQL +psql -d crelay -c "\COPY events FROM 'events.csv' WITH CSV" + +# 3. Migrate config +sqlite3 old_relay.db ".mode csv" "SELECT key,value,data_type,description,category,requires_restart FROM config" > config.csv +psql -d crelay -c "\COPY config(key,value,data_type,description,category,requires_restart) FROM 'config.csv' WITH CSV" + +# 4. Migrate auth rules +sqlite3 old_relay.db ".mode csv" "SELECT rule_type,pattern_type,pattern_value,active FROM auth_rules" > auth_rules.csv +psql -d crelay -c "\COPY auth_rules(rule_type,pattern_type,pattern_value,active) FROM 'auth_rules.csv' WITH CSV" + +# 5. Refresh materialized views +psql -d crelay -c "REFRESH MATERIALIZED VIEW event_kinds_mv; REFRESH MATERIALIZED VIEW time_stats_mv; REFRESH MATERIALIZED VIEW top_pubkeys_mv;" +``` + +--- + +## Phase 2: Multi-Instance + Dashboard + +### Goal + +Run multiple c-relay-pg instances behind a load balancer with a separate dashboard web server, all sharing the same PostgreSQL database. + +### Components + +1. **nginx config** — WebSocket load balancing with `ip_hash` stickiness +2. **systemd template** — `c-relay-pg@.service` for multiple instances +3. **LISTEN/NOTIFY integration** — Cross-instance event broadcasting in the `lws_service()` loop +4. **PgBouncer** — Connection pooling between c-relay-pg instances and PostgreSQL +5. **Dashboard** — Separate web server querying PostgreSQL directly +6. **Materialized view refresh** — Background job to refresh analytics views + +### Multi-Instance Architecture + +```mermaid +flowchart TD + INTERNET[Internet - Nostr Clients] -->|wss://relay.example.com| NGINX[nginx reverse proxy - TLS termination + load balancing] + + NGINX -->|ws://localhost:8888| R1[c-relay-pg Instance 1 - port 8888] + NGINX -->|ws://localhost:8889| R2[c-relay-pg Instance 2 - port 8889] + NGINX -->|ws://localhost:8890| R3[c-relay-pg Instance 3 - port 8890] + NGINX -->|http://localhost:3000| DASHWEB[Dashboard Web Server] + + R1 -->|libpq connection pool| PGPOOL[PgBouncer - connection pooler] + R2 -->|libpq connection pool| PGPOOL + R3 -->|libpq connection pool| PGPOOL + + PGPOOL -->|pooled connections| PG[PostgreSQL Primary] + DASHWEB -->|read queries| PG + + PG -->|streaming replication| REPLICA[Read Replica - optional] + DASHWEB -.->|heavy analytics| REPLICA +``` + +### nginx Load Balancing Config + +```nginx +# nginx.conf - WebSocket load balancing for c-relay-pg +upstream relay_backends { + # ip_hash ensures a client always hits the same instance + # (important for WebSocket session stickiness) + ip_hash; + + server 127.0.0.1:8888; + server 127.0.0.1:8889; + server 127.0.0.1:8890; +} + +server { + listen 443 ssl; + server_name relay.example.com; + + # TLS config... + + location / { + proxy_pass http://relay_backends; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_set_header X-Real-IP $remote_addr; + proxy_read_timeout 86400; # Keep WebSocket alive for 24h + } + + # Dashboard on separate path + location /dashboard { + proxy_pass http://127.0.0.1:3000; + } +} +``` + +**Session stickiness** (`ip_hash`) ensures that once a client connects to Instance 2, all their subsequent WebSocket frames go to Instance 2. This is important because subscriptions are held in-memory per instance. + +### Starting Multiple Instances + +Each instance is the same binary, just on a different port, all pointing to the same PostgreSQL: + +```bash +# Instance 1 +./build/c_relay_x86 --port 8888 --db-host localhost --db-name crelay & + +# Instance 2 +./build/c_relay_x86 --port 8889 --db-host localhost --db-name crelay & + +# Instance 3 +./build/c_relay_x86 --port 8890 --db-host localhost --db-name crelay & +``` + +Or with systemd template units: + +```ini +# /etc/systemd/system/c-relay-pg@.service +[Unit] +Description=C-Relay-PG Nostr Instance %i +After=postgresql.service + +[Service] +ExecStart=/opt/c-relay-pg/c_relay_x86 --port %i --db-host localhost --db-name crelay +Restart=always +User=c-relay + +[Install] +WantedBy=multi-user.target +``` + +```bash +systemctl enable c-relay-pg@8888 c-relay-pg@8889 c-relay-pg@8890 +systemctl start c-relay-pg@8888 c-relay-pg@8889 c-relay-pg@8890 +``` + +### Cross-Instance Event Broadcasting + +When Instance 1 receives a new EVENT and stores it in PostgreSQL, Instance 2 and Instance 3 need to know about it so they can broadcast to their connected subscribers. + +```mermaid +sequenceDiagram + participant Client_A as Client A - connected to Instance 1 + participant I1 as Instance 1 + participant PG as PostgreSQL + participant I2 as Instance 2 + participant I3 as Instance 3 + participant Client_B as Client B - connected to Instance 2 + participant Client_C as Client C - connected to Instance 3 + + Client_A->>I1: EVENT - new kind:1 note + I1->>PG: INSERT INTO events... + PG-->>I1: OK + I1->>I1: broadcast to local subscribers + I1->>PG: NOTIFY new_event with event_id + + Note over PG: PostgreSQL delivers notification to all listeners + + PG-->>I2: NOTIFY: new_event event_id + PG-->>I3: NOTIFY: new_event event_id + + I2->>PG: SELECT event_json FROM events WHERE id = event_id + PG-->>I2: event JSON + I2->>I2: match against local subscriptions + I2->>Client_B: EVENT message - if subscription matches + + I3->>PG: SELECT event_json FROM events WHERE id = event_id + PG-->>I3: event JSON + I3->>I3: match against local subscriptions + I3->>Client_C: EVENT message - if subscription matches +``` + +#### PostgreSQL LISTEN/NOTIFY Implementation + +Built into PostgreSQL — no additional infrastructure needed: + +```c +// === In the relay's event loop (modified lws_service loop) === + +// Setup: create a dedicated connection for LISTEN +PGconn* notify_conn = PQconnectdb("host=localhost dbname=crelay"); +PQexec(notify_conn, "LISTEN new_event"); +int notify_fd = PQsocket(notify_conn); // Get the socket fd for poll() + +// After storing an event: +void on_event_stored(PGconn* write_conn, const char* event_id) { + char notify_cmd[128]; + snprintf(notify_cmd, sizeof(notify_cmd), + "NOTIFY new_event, '%s'", event_id); + PQexec(write_conn, notify_cmd); +} + +// In the main event loop (runs every lws_service iteration): +void check_cross_instance_events(PGconn* notify_conn) { + // Non-blocking check for notifications + PQconsumeInput(notify_conn); + + PGnotify* notify; + while ((notify = PQnotifies(notify_conn)) != NULL) { + // Another instance stored a new event + const char* event_id = notify->extra; + + // Fetch the event and check against local subscriptions + cJSON* event = db_get_event_by_id(event_id); + if (event) { + broadcast_event_to_subscriptions(event); + cJSON_Delete(event); + } + + PQfreemem(notify); + } +} +``` + +**Performance**: LISTEN/NOTIFY adds ~1–5ms latency for cross-instance delivery. For a Nostr relay, this is imperceptible — clients already expect network latency. + +**Payload limit**: NOTIFY payloads are limited to 8000 bytes. For event IDs (64 hex chars), this is fine. + +#### Alternative: Redis Pub/Sub + +If you later need even lower latency or more sophisticated routing: + +```c +// Using hiredis (Redis C client) +redisContext* redis = redisConnect("127.0.0.1", 6379); + +// After storing event: +redisCommand(redis, "PUBLISH new_event %s", event_json); + +// Subscriber (in each instance): +redisCommand(redis, "SUBSCRIBE new_event"); +// Then poll for messages in the event loop +``` + +Redis adds sub-millisecond pub/sub but requires running a Redis server. For most relays, PostgreSQL LISTEN/NOTIFY is sufficient. + +### Connection Pooling with PgBouncer + +Each relay instance needs multiple database connections. Without pooling, 3 instances × 10 connections = 30 PostgreSQL backend processes. With PgBouncer: + +```ini +# /etc/pgbouncer/pgbouncer.ini +[databases] +crelay = host=127.0.0.1 port=5432 dbname=crelay + +[pgbouncer] +listen_port = 6432 +listen_addr = 127.0.0.1 +auth_type = md5 +pool_mode = transaction # Return connection to pool after each transaction +max_client_conn = 200 # Total connections from all relay instances +default_pool_size = 20 # Actual PostgreSQL connections +``` + +Relay instances connect to PgBouncer (port 6432) instead of PostgreSQL directly (port 5432). PgBouncer multiplexes 200 client connections onto 20 actual PostgreSQL connections. + +### Zero-Downtime Deployment + +```mermaid +sequenceDiagram + participant LB as nginx + participant I1 as Instance 1 - v1.0 + participant I2 as Instance 2 - v1.0 + participant I3 as Instance 3 - v1.0 + participant I1_NEW as Instance 1 - v1.1 + + Note over LB,I3: Normal operation: 3 instances serving traffic + + LB->>I1: Mark upstream as down + Note over I1: Drain: wait for existing connections to close or timeout + I1->>I1: Graceful shutdown + + Note over LB: Traffic now goes to I2 and I3 only + + I1_NEW->>I1_NEW: Start with new binary + I1_NEW->>LB: Health check passes + LB->>I1_NEW: Mark upstream as up + + Note over LB,I1_NEW: Instance 1 now running v1.1, repeat for I2 and I3 +``` + +Rolling deploy script: + +```bash +#!/bin/bash +# rolling_deploy.sh - Zero-downtime deployment + +for port in 8888 8889 8890; do + echo "Deploying instance on port $port..." + + # 1. Tell nginx to stop sending new connections + sed -i "s/server 127.0.0.1:$port;/server 127.0.0.1:$port down;/" /etc/nginx/nginx.conf + nginx -s reload + + # 2. Wait for existing connections to drain (30 seconds) + sleep 30 + + # 3. Stop old instance + systemctl stop c-relay-pg@$port + + # 4. Deploy new binary + cp ./build/c_relay_x86 /opt/c-relay-pg/c_relay_x86 + + # 5. Start new instance + systemctl start c-relay-pg@$port + + # 6. Wait for health check + sleep 5 + + # 7. Re-enable in nginx + sed -i "s/server 127.0.0.1:$port down;/server 127.0.0.1:$port;/" /etc/nginx/nginx.conf + nginx -s reload + + echo "Instance on port $port deployed successfully" +done +``` + +### Dashboard Options + +With PostgreSQL, the dashboard can be built with any technology: + +- **Grafana** — Point directly at PostgreSQL, zero custom code +- **Custom web app** — React/Vue frontend + any backend (Node, Python, Go) querying PostgreSQL +- **Embedded in c-relay-pg** — Keep current approach but queries go through `db_ops` to PostgreSQL (no longer blocks event loop since PostgreSQL handles concurrency) + +### Scaling Scenarios + +| Scenario | Instances | Connections | Events/hour | Setup | +|----------|-----------|-------------|-------------|-------| +| **Current** | 1 | ~1,200 | 56 | Single process + SQLite | +| **Small upgrade** | 2 | ~2,500 | 500 | 2 instances + PostgreSQL | +| **Medium relay** | 4 | ~5,000 | 5,000 | 4 instances + PostgreSQL + PgBouncer | +| **Large relay** | 8 | ~10,000 | 50,000 | 8 instances + PostgreSQL + read replica | +| **Multi-region** | 2–4 per region | ~50,000 | 500,000 | Multiple servers + PostgreSQL replication | + +### Cost Perspective + +Running multiple relay instances with PostgreSQL on a single VPS: +- **PostgreSQL**: ~200–500 MB RAM baseline +- **PgBouncer**: ~10 MB RAM +- **Each relay instance**: ~50–100 MB RAM +- **Dashboard web server**: ~50–100 MB RAM +- **4 instances + PostgreSQL + PgBouncer + dashboard**: ~1–2 GB total RAM +- A **$20/month VPS** with 4 cores and 4 GB RAM handles this easily +- A **$40/month VPS** with 8 cores and 8 GB RAM handles 8 instances comfortably + +--- + +## Risk Mitigation + +| Risk | Mitigation | +|------|-----------| +| PostgreSQL connection failures | `db_ops` returns error codes. Relay logs errors but doesn't crash. Reconnection logic with exponential backoff. | +| Performance regression | Benchmark before/after. PostgreSQL should be faster for complex queries, similar for simple ones. | +| Data loss during migration | Migration tool is read-only on SQLite. PostgreSQL import is idempotent (INSERT ON CONFLICT). | +| SQLite fallback needed | Keep `db_ops_sqlite.c` working. Compile-time flag switches backends. Can always go back. | +| LISTEN/NOTIFY message loss | NOTIFY is transactional — if the INSERT commits, the NOTIFY is guaranteed. Missed notifications during reconnect handled by periodic full-sync. | +| PgBouncer adds latency | Transaction pooling mode adds <0.1ms. Negligible compared to query time. | + +--- + +## Relationship to c-relay + +This project depends on the `db_ops.h` abstraction layer built in [c-relay thread pool plan](c_relay_thread_pool_plan.md) Phase 1. The interface is identical — only the backend implementation changes: + +- **c-relay**: `db_ops.c` → SQLite calls via `sqlite3_*` +- **c-relay-pg**: `db_ops_postgres.c` → PostgreSQL calls via `libpq` (`PQexec`, `PQgetvalue`, etc.) + +The `db_ops.h` header file is shared between both projects. Changes to the interface should be coordinated. diff --git a/plans/c_relay_thread_pool_plan.md b/plans/c_relay_thread_pool_plan.md new file mode 100644 index 0000000..cb4f7ce --- /dev/null +++ b/plans/c_relay_thread_pool_plan.md @@ -0,0 +1,332 @@ +# c-relay Thread Pool Plan — db_ops Abstraction + SQLite Thread Pool + +## Overview + +This plan covers improvements to **c-relay** (this repo) — the lean, single-binary, embedded-SQLite Nostr relay. Two phases: + +1. **Phase 1: Database Abstraction Layer** — Consolidate all scattered `sqlite3_*` calls into a single `db_ops.h` / `db_ops.c` module. Pure refactor, no behavior change. +2. **Phase 2: SQLite Thread Pool** — Add worker threads for concurrent reads, unblocking the `lws_service()` event loop. + +This is the **final form of c-relay**: an embedded-SQLite relay with clean architecture and non-blocking database access. The abstraction layer also enables a future fork to PostgreSQL — see [c-relay-pg plan](c_relay_pg_plan.md). + +### Why This First? + +The current architecture has a fundamental bottleneck documented in the [database architecture analysis](database_architecture_analysis.md): the single-threaded event loop blocks on every database call. A 672ms REQ query freezes every connected client. The thread pool fixes this without changing the database engine or deployment model. + +| Metric | Current | After Thread Pool | +|--------|---------|-------------------| +| **Event loop blocking** | 0.1–672ms per query | Near-zero | +| **Concurrent reads** | 1 | 4–8 | +| **Deployment** | Single binary + DB file | Same | +| **Database** | SQLite | Same | +| **Code risk** | N/A | Low — additive change | + +## Architecture + +```mermaid +flowchart TD + subgraph c-relay - Final Form + RELAY[c-relay binary] -->|db_ops API| DBOPS[db_ops.c - abstraction layer] + DBOPS -->|sqlite3 calls| SQLITE[SQLite WAL database] + end +``` + +```mermaid +flowchart TD + subgraph c-relay with Thread Pool + LWS[lws_service event loop] -->|REQ arrives| Q[Thread-safe job queue] + LWS -->|EVENT arrives| WQ[Write queue - single writer] + Q --> T1[Reader Thread 1 - own sqlite3*] + Q --> T2[Reader Thread 2 - own sqlite3*] + Q --> T3[Reader Thread 3 - own sqlite3*] + Q --> T4[Reader Thread 4 - own sqlite3*] + WQ --> TW[Writer Thread - own sqlite3*] + T1 -->|results| CB[Callback to lws event loop] + T2 -->|results| CB + T3 -->|results| CB + T4 -->|results| CB + TW -->|OK/error| CB + CB -->|queue_message| LWS + end +``` + +--- + +## Phase 1: Database Abstraction Layer + +### Goal + +Consolidate all 258 scattered `sqlite3_*` calls across 8 files into a single `db_ops.h` / `db_ops.c` module with a backend-agnostic interface. SQLite continues to work — this is a pure refactor. + +### Files That Currently Touch SQLite Directly + +| File | `sqlite3_*` calls | Operations | +|------|-------------------|------------| +| [`src/main.c`](../src/main.c) | ~80 | Event store/retrieve, tag storage, REQ queries, COUNT queries, DB init, schema migration | +| [`src/config.c`](../src/config.c) | ~60 | Config CRUD, auth rules CRUD, WoT sync, relay key storage, startup sequence | +| [`src/api.c`](../src/api.c) | ~40 | Stats queries, monitoring views, admin SQL execution, config management | +| [`src/dm_admin.c`](../src/dm_admin.c) | ~20 | Auth rule management, WoT whitelist operations | +| [`src/websockets.c`](../src/websockets.c) | ~15 | COUNT queries, IP ban stats, connection tracking | +| [`src/subscriptions.c`](../src/subscriptions.c) | ~15 | Subscription logging — create/close/disconnect | +| [`src/nip009.c`](../src/nip009.c) | ~10 | Event deletion — by ID and by address | +| [`src/request_validator.c`](../src/request_validator.c) | ~8 | Blacklist/whitelist checks | +| [`src/ip_ban.c`](../src/ip_ban.c) | ~15 | IP ban table CRUD, persistence | + +### Abstraction Layer Design + +New files: `src/db_ops.h` and `src/db_ops.c` + +```c +// src/db_ops.h — Database operations abstraction layer +#ifndef DB_OPS_H +#define DB_OPS_H + +#include "../nostr_core_lib/cjson/cJSON.h" + +// ============================================================ +// Lifecycle +// ============================================================ +int db_init(const char* connection_string); // SQLite: file path, PG: connection string +void db_close(void); +int db_is_available(void); + +// ============================================================ +// Schema / Migration +// ============================================================ +int db_ensure_schema(void); +int db_get_schema_version(void); +int db_execute_raw(const char* sql); // For schema DDL only + +// ============================================================ +// Event Operations +// ============================================================ +int db_store_event(cJSON* event); +int db_event_exists(const char* event_id); +char* db_get_event_json(const char* event_id); // Caller must free +int db_delete_event_by_id(const char* event_id, const char* requester_pubkey); +int db_delete_events_by_address(const char* pubkey, int kind, + const char* d_tag, long before_timestamp); + +// ============================================================ +// Event Queries - REQ handling +// ============================================================ +typedef struct { + int* kinds; int kind_count; + char** authors; int author_count; + char** ids; int id_count; + char** tag_names; // Parallel arrays: tag_names[i] has tag_values[i][] + char*** tag_values; + int* tag_value_counts; + int tag_filter_count; + long since; // 0 = not set + long until; // 0 = not set + int limit; // 0 = default 500 + char* search; // NIP-50 search term, NULL = not set +} db_event_filter_t; + +typedef struct db_result db_result_t; + +db_result_t* db_query_events(const db_event_filter_t* filter); +const char* db_result_next_json(db_result_t* result); // Returns event_json, NULL when done +int db_result_row_count(db_result_t* result); +void db_result_free(db_result_t* result); + +int db_count_events(const db_event_filter_t* filter); + +// ============================================================ +// Config Operations +// ============================================================ +char* db_get_config(const char* key); // Caller must free, NULL if not found +int db_set_config(const char* key, const char* value, const char* data_type, + const char* description, const char* category, int requires_restart); +int db_update_config(const char* key, const char* value); +int db_config_exists(const char* key); +int db_get_config_count(void); + +// ============================================================ +// Auth Rules Operations +// ============================================================ +int db_add_auth_rule(const char* rule_type, const char* pattern_type, const char* pattern_value); +int db_remove_auth_rule(const char* rule_type, const char* pattern_type, const char* pattern_value); +int db_auth_rule_exists(const char* rule_type, const char* pattern_type, const char* pattern_value); +int db_clear_auth_rules(const char* rule_type); // NULL = clear all +int db_is_blacklisted(const char* pubkey); +int db_is_whitelisted(const char* pubkey); +int db_has_any_whitelist(void); + +// ============================================================ +// Relay Key Storage +// ============================================================ +int db_store_relay_private_key(const char* privkey_hex); +char* db_get_relay_private_key(void); // Caller must free + +// ============================================================ +// Subscription Logging +// ============================================================ +int db_log_subscription_created(const char* sub_id, const char* wsi_ptr, + const char* client_ip, const char* filter_json); +int db_log_subscription_closed(const char* sub_id, const char* client_ip); +int db_log_subscription_disconnected(const char* client_ip); +int db_update_subscription_events_sent(const char* sub_id, int events_sent); +int db_cleanup_orphaned_subscriptions(void); + +// ============================================================ +// IP Ban Persistence +// ============================================================ +int db_ensure_ip_ban_table(void); +int db_load_ip_bans(void* ban_table, int table_size); // Populates in-memory table +int db_save_ip_bans(const void* ban_table, int table_size); + +// ============================================================ +// Analytics / Stats - for dashboard +// ============================================================ +cJSON* db_get_event_kind_distribution(void); +cJSON* db_get_time_based_stats(void); +cJSON* db_get_top_pubkeys(int limit); +cJSON* db_get_subscription_details(void); +int db_get_total_event_count(void); + +// ============================================================ +// Admin SQL Query +// ============================================================ +cJSON* db_execute_admin_query(const char* sql, char* error_msg, size_t error_size); + +#endif // DB_OPS_H +``` + +### Migration Strategy for Phase 1 + +The key principle: **change the interface, not the behavior**. Each function in `db_ops.c` initially just wraps the existing SQLite calls. + +**Step-by-step for each file:** + +1. **Create `db_ops.h` and `db_ops.c`** with the interface above +2. **Implement each `db_ops` function** by moving the existing SQLite code from the source files into `db_ops.c` +3. **Replace direct `sqlite3_*` calls** in each source file with `db_ops_*` calls +4. **Remove `extern sqlite3* g_db`** from each file — only `db_ops.c` knows about `g_db` +5. **Remove `#include `** from each file — only `db_ops.c` includes it +6. **Update Makefile** to compile `db_ops.c` + +### Order of Migration (by risk, lowest first) + +1. **`src/ip_ban.c`** — Self-contained, simple CRUD. Good warmup. +2. **`src/subscriptions.c`** — Logging only, no critical path. +3. **`src/nip009.c`** — Event deletion, small file. +4. **`src/request_validator.c`** — Auth checks, small file. +5. **`src/api.c`** — Stats/monitoring queries. Larger but read-only. +6. **`src/dm_admin.c`** — Auth rules + WoT. Medium complexity. +7. **`src/config.c`** — Config CRUD. Large but well-structured. +8. **`src/websockets.c`** — COUNT queries, minimal DB usage. +9. **`src/main.c`** — Event store/retrieve, REQ queries. The big one — do last. + +### Testing Phase 1 + +After Phase 1, the relay should behave **identically** to before. Run: +- `tests/run_all_tests.sh` — Full test suite +- `tests/performance_benchmarks.sh` — Verify no performance regression +- Manual testing with production-like traffic + +--- + +## Phase 2: SQLite Thread Pool + +### Goal + +Add a pool of N worker threads, each with its own `sqlite3*` connection to the same database file. SQLite WAL mode (already enabled) supports **concurrent readers**. The event loop dispatches database work to the pool and remains responsive. + +### Key Design Points + +1. **Read path**: REQ queries dispatched to thread pool. Each worker opens its own `sqlite3*` connection. SQLite WAL allows unlimited concurrent readers. + +2. **Write path**: EVENT inserts go through a single dedicated writer thread. SQLite only allows one writer at a time anyway — this serializes writes cleanly. + +3. **Result delivery**: Worker threads cannot call `lws_write()` directly (libwebsockets is not thread-safe). Instead, they push results into a per-session message queue and call `lws_cancel_service()` to wake the event loop, which then drains the queue. + +4. **Connection lifecycle**: Each thread opens its own connection with `PRAGMA journal_mode=WAL` and `PRAGMA busy_timeout=5000`. + +### What Changes in the Codebase + +| Component | Current | Thread Pool | +|-----------|---------|-------------| +| [`g_db`](../src/main.c:49) | Single global connection | One per thread + writer connection | +| [`handle_req_message()`](../src/main.c:1101) | Synchronous SQL in callback | Package filter into job, dispatch to pool | +| [`store_event()`](../src/main.c:787) | Synchronous INSERT in callback | Dispatch to writer thread | +| [`handle_count_message()`](../src/websockets.c:2863) | Synchronous COUNT in callback | Dispatch to pool | +| Result delivery | Direct `queue_message()` | Worker pushes to queue + `lws_cancel_service()` | +| Config reads | Direct `sqlite3_prepare` on `g_db` | Can stay synchronous with own connection or cache | + +### New Files + +- `src/thread_pool.h` — Thread pool interface +- `src/thread_pool.c` — Thread pool implementation (job queue, worker lifecycle, result delivery) + +### Thread Pool Interface (sketch) + +```c +// src/thread_pool.h +#ifndef THREAD_POOL_H +#define THREAD_POOL_H + +typedef enum { + JOB_TYPE_REQ_QUERY, + JOB_TYPE_COUNT_QUERY, + JOB_TYPE_STORE_EVENT, + JOB_TYPE_DELETE_EVENT, +} job_type_t; + +typedef struct { + job_type_t type; + void* session; // lws per-session data pointer + db_event_filter_t filter; // For REQ/COUNT jobs + cJSON* event; // For STORE jobs + char event_id[65]; // For DELETE jobs +} db_job_t; + +int thread_pool_init(int num_readers, const char* db_path); +void thread_pool_shutdown(void); +int thread_pool_submit_read(db_job_t* job); +int thread_pool_submit_write(db_job_t* job); + +#endif // THREAD_POOL_H +``` + +### Performance Characteristics + +| Metric | Value | +|--------|-------| +| **Concurrent reads** | N readers in parallel (N = thread count, typically 4–8) | +| **Write throughput** | Same as current — SQLite serializes writes regardless | +| **Event loop latency** | Near-zero — REQ no longer blocks the loop | +| **Max theoretical read throughput** | ~4–8x current (limited by disk I/O, not CPU) | +| **Memory overhead** | ~50–100 MB per connection (page cache) | +| **Latency per query** | Same as current per-query, but no head-of-line blocking | + +### Limitations + +- **Write contention**: SQLite still allows only ONE writer at a time. With WAL, readers don't block writers and writers don't block readers, but two simultaneous writes will serialize. At the current write rate (~56 events/hour), this is a non-issue. +- **Database size**: The 2.7 GB index bloat problem remains. Thread pool doesn't fix the schema — it fixes the concurrency. +- **Scaling ceiling**: Beyond ~8 reader threads, diminishing returns due to disk I/O contention on a single SQLite file. +- **Complexity**: Need a proper job queue, thread lifecycle management, and careful handling of the lws ↔ worker thread boundary. + +--- + +## Risk Mitigation + +| Risk | Mitigation | +|------|-----------| +| Phase 1 introduces bugs | Each file migrated independently, tested after each. Full test suite runs after each file. | +| Thread pool race conditions | Worker threads only touch their own `sqlite3*` connection. Result delivery uses a mutex-protected queue. `lws_cancel_service()` is documented as thread-safe. | +| Performance regression from abstraction | Abstraction layer is thin wrappers — no extra allocations or copies. Benchmark before/after. | +| lws thread safety issues | Workers never call `lws_write()` directly. They push to a queue and wake the event loop. This is the documented pattern for libwebsockets multi-threading. | +| Rollback needed | Phase 1 is a pure refactor — easy to revert. Phase 2 thread pool is additive — can be disabled with a compile flag. | + +--- + +## Relationship to c-relay-pg + +After Phase 1 (abstraction layer) is complete, the codebase is ready to be forked into a separate **c-relay-pg** project that replaces the SQLite backend with PostgreSQL. See [c-relay-pg plan](c_relay_pg_plan.md) for details. + +The `db_ops.h` interface is designed to work with any backend: +- **SQLite** (this plan): `db_result_next_json()` copies from `sqlite3_column_text()` +- **PostgreSQL** (c-relay-pg): `db_result_next_json()` returns from `PQgetvalue()` +- **LMDB** (future possibility): `db_result_next_json()` returns a zero-copy pointer into mmap'd memory diff --git a/plans/database_architecture_analysis.md b/plans/database_architecture_analysis.md new file mode 100644 index 0000000..2e9c6c1 --- /dev/null +++ b/plans/database_architecture_analysis.md @@ -0,0 +1,1055 @@ +# Database Architecture Analysis: Thread Pool vs Client-Server DB vs LMDB + +## Current Architecture Problem + +The c-relay runs a **single-threaded event loop** via `lws_service()` in [`start_websocket_relay()`](src/websockets.c:2637). Every WebSocket callback — including [`handle_req_message()`](src/main.c:1101) and [`store_event()`](src/main.c:787) — executes **synchronously** inside this loop. When a REQ query takes 672ms (as documented in your [query analysis report](query_analysis_report.md:39)), **every other connected client is frozen** for that duration. + +### The Blocking Chain + +```mermaid +sequenceDiagram + participant C1 as Client 1 + participant LWS as lws_service loop + participant DB as SQLite g_db + participant C2 as Client 2 + + C1->>LWS: REQ with tag filter + LWS->>DB: sqlite3_prepare + step loop + Note over LWS,DB: BLOCKED 50-672ms + C2->>LWS: EVENT submission + Note over C2: Waiting... event loop frozen + DB-->>LWS: Results returned + LWS-->>C1: EVENT responses + EOSE + LWS->>DB: store_event for C2 + Note over LWS,DB: BLOCKED again for write + DB-->>LWS: Write complete + LWS-->>C2: OK response +``` + +### What the Numbers Tell Us + +From your production data: +- **2.7 GB database** but only **186 MB actual data** — 98% is indexes and overhead +- **4 million `event_tags` rows** with 3 indexes = ~2 GB of index space +- **Average query: 10.4ms**, worst case: **672ms** +- **98% of queries are REQ reads**, only 2% are writes +- **~120 new subscriptions/minute** = 2 queries/second minimum hitting the DB + +The single `sqlite3* g_db` connection is shared across all operations with **no mutex protection on the connection itself** — it works only because everything runs on one thread. This is the fundamental bottleneck. + +--- + +## Option 2: SQLite Thread Pool + +### How It Works + +Create a pool of N worker threads, each with its own `sqlite3*` connection to the same database file. SQLite WAL mode (already enabled) supports **concurrent readers**. The architecture becomes: + +```mermaid +flowchart TD + LWS[lws_service event loop] -->|REQ arrives| Q[Thread-safe job queue] + LWS -->|EVENT arrives| WQ[Write queue - single writer] + Q --> T1[Reader Thread 1 - own sqlite3*] + Q --> T2[Reader Thread 2 - own sqlite3*] + Q --> T3[Reader Thread 3 - own sqlite3*] + Q --> T4[Reader Thread 4 - own sqlite3*] + WQ --> TW[Writer Thread - own sqlite3*] + T1 -->|results| CB[Callback to lws event loop] + T2 -->|results| CB + T3 -->|results| CB + T4 -->|results| CB + TW -->|OK/error| CB + CB -->|queue_message| LWS +``` + +### Key Design Points + +1. **Read path**: REQ queries dispatched to thread pool. Each worker opens its own `sqlite3*` connection. SQLite WAL allows unlimited concurrent readers. + +2. **Write path**: EVENT inserts go through a single dedicated writer thread. SQLite only allows one writer at a time anyway — this serializes writes cleanly. + +3. **Result delivery**: Worker threads cannot call `lws_write()` directly (libwebsockets is not thread-safe). Instead, they push results into a per-session message queue and call `lws_cancel_service()` to wake the event loop, which then drains the queue. + +4. **Connection lifecycle**: Each thread opens its own connection with `PRAGMA journal_mode=WAL` and `PRAGMA busy_timeout=5000`. + +### What Changes in the Codebase + +| Component | Current | Thread Pool | +|-----------|---------|-------------| +| [`g_db`](src/main.c:49) | Single global connection | One per thread + writer connection | +| [`handle_req_message()`](src/main.c:1101) | Synchronous SQL in callback | Package filter into job, dispatch to pool | +| [`store_event()`](src/main.c:787) | Synchronous INSERT in callback | Dispatch to writer thread | +| [`handle_count_message()`](src/websockets.c:2863) | Synchronous COUNT in callback | Dispatch to pool | +| Result delivery | Direct `queue_message()` | Worker pushes to queue + `lws_cancel_service()` | +| Config reads | Direct `sqlite3_prepare` on `g_db` | Can stay synchronous with own connection or cache | + +### Performance Characteristics + +| Metric | Value | +|--------|-------| +| **Concurrent reads** | N readers in parallel (N = thread count, typically 4-8) | +| **Write throughput** | Same as current — SQLite serializes writes regardless | +| **Event loop latency** | Near-zero — REQ no longer blocks the loop | +| **Max theoretical read throughput** | ~4-8x current (limited by disk I/O, not CPU) | +| **Memory overhead** | ~50-100 MB per connection (page cache) | +| **Latency per query** | Same as current per-query, but no head-of-line blocking | + +### Limitations + +- **Write contention**: SQLite still allows only ONE writer at a time. With WAL, readers don't block writers and writers don't block readers, but two simultaneous writes will serialize. At your write rate (~56 events/hour), this is a non-issue. +- **Database size**: The 2.7 GB index bloat problem remains. Thread pool doesn't fix the schema — it fixes the concurrency. +- **Scaling ceiling**: Beyond ~8 reader threads, you hit diminishing returns due to disk I/O contention on a single SQLite file. +- **Complexity**: Need a proper job queue, thread lifecycle management, and careful handling of the lws ↔ worker thread boundary. + +--- + +## Option 3: Client-Server Database (PostgreSQL) + +### How It Works + +Replace SQLite with PostgreSQL. The relay connects via `libpq` (PostgreSQL C client library). PostgreSQL runs as a separate process with its own connection pooling, query planner, and MVCC concurrency. + +```mermaid +flowchart TD + LWS[lws_service event loop] -->|async query| PG[libpq async connection pool] + PG -->|TCP/Unix socket| PGS[PostgreSQL Server] + PGS --> D1[Disk - WAL] + PGS --> D2[Shared Buffers - RAM cache] + PGS --> W1[Worker Process 1] + PGS --> W2[Worker Process 2] + PGS --> WN[Worker Process N] + W1 -->|results| PG + PG -->|callback| LWS +``` + +### Performance Characteristics + +| Metric | Value | +|--------|-------| +| **Concurrent reads** | Unlimited — each query gets its own backend process | +| **Concurrent writes** | True concurrent writes with row-level locking | +| **Event loop latency** | Near-zero with async `libpq` | +| **Max theoretical throughput** | 10-100x SQLite depending on hardware | +| **Memory overhead** | ~10 MB per PostgreSQL backend + shared_buffers | +| **Latency per query** | Slightly higher for simple queries due to TCP/IPC overhead, but much better for complex queries due to superior query planner | +| **Index efficiency** | Far superior — PostgreSQL B-tree indexes are more space-efficient, supports partial indexes, GIN indexes for JSON | + +### What Changes in the Codebase + +**Everything touching `sqlite3*` must be rewritten.** This is ~258 call sites across 8 files: + +| File | sqlite3 calls | Scope of change | +|------|--------------|-----------------| +| [`src/main.c`](src/main.c) | ~80 | Complete rewrite of store_event, handle_req_message, retrieve_event, init_database | +| [`src/config.c`](src/config.c) | ~60 | All config table operations | +| [`src/websockets.c`](src/websockets.c) | ~15 | COUNT queries, connection tracking | +| [`src/api.c`](src/api.c) | ~40 | All monitoring/stats queries | +| [`src/dm_admin.c`](src/dm_admin.c) | ~20 | Auth rules, WoT sync | +| [`src/subscriptions.c`](src/subscriptions.c) | ~15 | Subscription logging | +| [`src/nip009.c`](src/nip009.c) | ~10 | Event deletion | +| [`src/ip_ban.c`](src/ip_ban.c) | ~15 | IP ban persistence | +| [`src/request_validator.c`](src/request_validator.c) | ~8 | Auth rule checks | + +### Advantages Over Thread Pool + +1. **True write concurrency**: PostgreSQL handles concurrent writes with row-level locking. If your relay grows to thousands of events/hour, this matters. +2. **Superior query planner**: PostgreSQL's cost-based optimizer is far more sophisticated than SQLite's. Complex tag queries with JOINs will be faster. +3. **Index efficiency**: PostgreSQL's indexes are more compact. Your 2 GB of SQLite indexes would likely be ~500 MB in PostgreSQL. GIN indexes on JSONB would eliminate the `event_tags` table entirely. +4. **JSONB native**: PostgreSQL has native JSONB with indexing. You could store tags as JSONB and query them directly with `@>` operator — no denormalized `event_tags` table needed. +5. **Connection pooling**: PgBouncer or built-in pooling handles thousands of concurrent connections efficiently. +6. **Operational tooling**: `pg_stat_statements`, `EXPLAIN ANALYZE`, `pg_dump`, replication, etc. +7. **Horizontal scaling**: Read replicas possible for future growth. + +### Disadvantages vs Thread Pool + +1. **Massive rewrite**: ~258 call sites across 8 files. This is essentially rewriting the entire data layer. +2. **External dependency**: PostgreSQL must be installed, configured, and maintained. SQLite is zero-config embedded. +3. **Deployment complexity**: Your current deployment is a single binary + SQLite file. PostgreSQL adds a service dependency. +4. **Latency for simple queries**: A simple `SELECT 1 FROM events WHERE id=?` is ~0.1ms in SQLite vs ~0.5ms in PostgreSQL due to IPC overhead. For your 98% read workload, this adds up. +5. **Memory footprint**: PostgreSQL server uses 100-500 MB baseline. SQLite uses ~50 MB. +6. **Testing complexity**: Tests need a running PostgreSQL instance. + +--- + +## Option 4: LMDB (Lightning Memory-Mapped Database) + +### What Is LMDB? + +LMDB is an embedded key-value store created by Howard Chu for OpenLDAP. It's what **strfry** — the fastest known Nostr relay — uses as its storage engine. It's fundamentally different from both SQLite and PostgreSQL: + +- **Memory-mapped**: The entire database is `mmap()`'d into the process address space. Reads are literally pointer dereferences — no syscalls, no copies, no serialization. +- **B+ tree on disk**: Data is stored in a B+ tree that maps directly to memory pages. The OS page cache IS the database cache. +- **MVCC with copy-on-write**: Readers never block writers, writers never block readers. No locks needed for reads at all. +- **Zero-copy reads**: When you read a value, you get a pointer directly into the mmap'd region. No `malloc`, no `memcpy`. +- **Single-writer, multiple-reader**: Like SQLite WAL, but implemented at a much lower level with near-zero overhead. + +### How It Would Work for c-relay + +LMDB is a **key-value store**, not a relational database. There's no SQL. You design your own indexes as separate "databases" (sub-B-trees within the same file): + +```mermaid +flowchart TD + subgraph LMDB Environment + DB1[events_by_id: event_id -> event_json] + DB2[events_by_pubkey: pubkey+created_at -> event_id] + DB3[events_by_kind: kind+created_at -> event_id] + DB4[events_by_tag: tag_name+tag_value -> event_id] + DB5[events_by_time: created_at -> event_id] + DB6[config: key -> value] + DB7[auth_rules: rule_id -> rule_data] + end + + REQ[REQ query] -->|lookup| DB3 + REQ -->|lookup| DB4 + DB3 -->|get event_id| DB1 + DB4 -->|get event_id| DB1 + DB1 -->|zero-copy pointer| Response[Send to client] +``` + +### The Key Insight: Why strfry Is Fast + +strfry doesn't use SQL at all. When a REQ comes in with `kinds: [1], #p: [pubkey123]`, strfry: + +1. Opens a read-only LMDB transaction (no locks, no copies) +2. Seeks to `tag_p:pubkey123` in the tag index B-tree +3. Iterates matching event IDs +4. For each ID, does a direct pointer lookup in the events B-tree +5. Returns the raw bytes — zero-copy, no JSON parsing, no serialization + +Compare this to your current c-relay flow: +1. Build SQL string with parameter binding +2. `sqlite3_prepare_v2()` — parse SQL, build query plan +3. `sqlite3_step()` — traverse B-tree, copy data to SQLite's page cache, then copy to your buffer +4. `sqlite3_column_text()` — copy string out of SQLite's internal format +5. `cJSON_Parse()` — parse the JSON string back into objects (for expiration check) +6. Build EVENT message string +7. Queue for sending + +Steps 2-5 are **completely eliminated** with LMDB. The event JSON bytes go straight from the mmap'd file to the WebSocket send buffer. + +### Performance Characteristics + +| Metric | SQLite (current) | SQLite Thread Pool | PostgreSQL | LMDB | +|--------|-----------------|-------------------|------------|------| +| **Read latency** | 0.1-672ms | Same per-query | 0.5-50ms | **0.001-0.1ms** | +| **Concurrent reads** | 1 (single thread) | 4-8 | Unlimited | **Unlimited** (lock-free) | +| **Write latency** | 1-10ms | Same | 0.5-5ms | **0.01-1ms** | +| **Concurrent writes** | 1 | 1 | Many | **1** (single writer) | +| **Memory copies per read** | 3-4 | 3-4 | 2-3 | **0** (zero-copy) | +| **Index overhead** | ~2 GB for 186 MB data | Same | ~500 MB | **~200-400 MB** | +| **CPU per query** | High (SQL parse + JSON parse) | Same | Medium (SQL parse) | **Minimal** (pointer math) | + +### What Changes in the Codebase + +This is the **largest rewrite** of all options because you're replacing SQL with manual index management: + +| Component | Current (SQLite) | LMDB Equivalent | +|-----------|-----------------|------------------| +| Schema definition | SQL DDL in [`sql_schema.h`](src/sql_schema.h) | C code defining named databases + key formats | +| [`store_event()`](src/main.c:787) | SQL INSERT with 9 bound params | `mdb_put()` into events db + `mdb_put()` into each index db | +| [`handle_req_message()`](src/main.c:1101) | Dynamic SQL builder (80 lines) | Manual cursor iteration across index databases with set intersection | +| Tag queries | `SELECT ... FROM event_tags WHERE tag_name=? AND tag_value IN (...)` | `mdb_cursor_get()` on tag index with `MDB_SET_RANGE` | +| Config reads | `SELECT value FROM config WHERE key=?` | `mdb_get()` on config database | +| Views/triggers | SQL views for analytics, triggers for replaceable events | Manual C code for all of it | +| Expiration check | `cJSON_Parse()` on every row | Can store expiration as separate indexed field — skip expired during cursor iteration | + +### Advantages + +1. **Raw speed**: 100-1000x faster reads than SQLite for your workload. strfry handles 10,000+ concurrent connections on modest hardware. +2. **Zero-copy**: Event JSON goes from disk → mmap → WebSocket buffer with no intermediate copies. +3. **No SQL overhead**: No query parsing, no query planning, no result materialization. +4. **Embedded**: Like SQLite, it's a library linked into your binary. No external server process. +5. **Proven for Nostr**: strfry demonstrates this works at scale for exactly this use case. +6. **Compact storage**: B+ tree is more space-efficient than SQLite's B-tree + WAL + journal overhead. +7. **Crash-safe**: MVCC with copy-on-write means the database is always consistent, even after power loss. +8. **No event loop blocking**: Read transactions are so fast (microseconds) they can run inline in the lws callback without needing a thread pool. + +### Disadvantages + +1. **No SQL**: You lose the ability to write ad-hoc queries. All query patterns must be pre-designed as index databases. Your admin SQL query API ([`api.c`](src/api.c) `execute_admin_sql_query()`) would need to be completely rethought. +2. **Manual index management**: Every query pattern needs its own index database. Adding a new filter type means adding a new index and backfilling it. +3. **Largest rewrite**: More code changes than even PostgreSQL, because you're replacing a query language with manual data structure operations. +4. **Single writer**: Like SQLite, only one write transaction at a time. Fine for your 56 events/hour, but a hard ceiling. +5. **No complex queries**: Queries like "top 10 pubkeys by event count" or "events per kind distribution" require manual aggregation in C code. Your monitoring views ([`event_kinds_view`](src/sql_schema.h:276), [`time_stats_view`](src/sql_schema.h:297)) would need C implementations. +6. **Memory mapping limits**: The database size is limited by virtual address space. On 64-bit systems this is effectively unlimited, but you must set `mapsize` at open time. +7. **Learning curve**: LMDB's API is low-level. Cursor management, transaction scoping, and key design require careful thought. +8. **Loss of admin SQL API**: Your current admin API supports arbitrary SQL queries via DM commands. This would be impossible with LMDB — you'd need to build specific query endpoints for each operation. + +### The strfry Precedent + +strfry's architecture is worth studying: +- Uses LMDB with custom indexes for each Nostr filter type +- Handles tag queries by maintaining a `tag_name:tag_value → event_id` index +- Supports NIP-01 filters by intersecting results from multiple index cursors +- Achieves sub-millisecond query times even with millions of events +- Single-threaded event loop (like your current design) but doesn't need a thread pool because reads are microseconds + +However, strfry is **purpose-built** around LMDB from day one. Retrofitting LMDB into an existing SQLite-based codebase is significantly harder than starting fresh. + +--- + +## Option 5: Document/JSON Databases + +Since Nostr events are JSON documents, document databases seem like a natural fit. Let's evaluate the relevant options. + +### CouchDB + +CouchDB stores JSON documents natively, uses HTTP as its protocol, and builds indexes via JavaScript map-reduce views. You have experience with it. + +**How it would work**: Each Nostr event becomes a CouchDB document. You'd create views for each query pattern — by kind, by pubkey, by tag, etc. CouchDB's MVCC model means readers never block writers. + +**Why it's NOT a good fit for a Nostr relay**: + +| Factor | Assessment | +|--------|-----------| +| **Latency** | HTTP API adds 1-5ms per request. Your current SQLite averages 10ms, so CouchDB wouldn't be faster for simple queries. | +| **C integration** | CouchDB is an Erlang server with an HTTP API. From C, every query is an HTTP request via libcurl. This is far heavier than a direct function call to SQLite or LMDB. | +| **View building** | Map-reduce views are written in JavaScript and built lazily. First query after data changes triggers a full view rebuild — this would cause massive latency spikes. | +| **Deployment** | Requires Erlang runtime + CouchDB server. Much heavier than PostgreSQL. | +| **Concurrency** | Good MVCC, but the HTTP overhead negates the benefit for an embedded relay. | +| **Real-time** | CouchDB's `_changes` feed could be useful for subscription broadcasting, but the HTTP polling model adds latency vs in-process callbacks. | + +**Verdict**: CouchDB is excellent for web applications where you're already using HTTP, but for a C relay that needs microsecond-level response times, the HTTP layer is a dealbreaker. + +### MongoDB + +MongoDB is the most popular document database. It stores BSON (binary JSON), has rich query operators, and supports secondary indexes on any field including nested JSON paths. + +**How it would work**: Events stored as BSON documents. Create indexes on `kind`, `pubkey`, `created_at`, and `tags` (using multikey indexes on arrays). Queries use MongoDB's query language which maps well to Nostr filters. + +**Why it's a mixed fit**: + +| Factor | Assessment | +|--------|-----------| +| **Query model** | Excellent — MongoDB's query operators map almost 1:1 to Nostr filters. `{kind: {$in: [1,7]}, tags: {$elemMatch: {0: "p", 1: "pubkey123"}}}` | +| **C driver** | `libmongoc` is mature and well-maintained. Supports async operations. | +| **Performance** | 0.5-5ms per query — faster than your current SQLite for complex queries, slower for simple ones. | +| **Index efficiency** | Multikey indexes on the tags array would eliminate the `event_tags` table problem. | +| **Deployment** | Requires MongoDB server. Heavy — 500 MB+ RAM baseline. | +| **Operational** | Needs replica set for durability. Single-node MongoDB is not crash-safe by default. | +| **Memory** | WiredTiger engine uses ~50% of RAM for cache by default. On a small VPS this is aggressive. | + +**Verdict**: MongoDB's query model is the best match for Nostr filters of any database, but the operational overhead is massive for a single relay. It's designed for clusters, not embedded use. + +### RethinkDB + +RethinkDB is a document database with built-in real-time push via "changefeeds" — when a document changes, subscribed queries automatically receive the update. + +**Why it's interesting for Nostr**: The changefeed model maps directly to Nostr subscriptions. When a new event is stored, RethinkDB could automatically push it to all matching subscriptions without your relay needing to do the [`broadcast_event_to_subscriptions()`](src/subscriptions.c:806) logic. + +**Why it's NOT practical**: +- RethinkDB development has slowed significantly (community-maintained since 2017) +- C driver is not officially supported +- Deployment complexity similar to MongoDB +- Query performance is slower than MongoDB for simple lookups + +**Verdict**: Interesting concept but not production-viable for a C relay. + +### UnQLite — The Embedded Document Store + +UnQLite is worth a closer look. It's an **embedded** key-value and document store — like SQLite but for JSON. Single C file, no external dependencies, public domain license. + +**How it would work**: + +```c +// Store event +unqlite_kv_store(pDb, event_id, 64, event_json, event_json_len); + +// Retrieve event by ID +unqlite_kv_fetch(pDb, event_id, 64, buffer, &buf_len); + +// For queries: use Jx9 scripting engine (embedded JavaScript-like language) +// Or: maintain manual indexes like LMDB +``` + +| Factor | Assessment | +|--------|-----------| +| **Integration** | Single C file, embeds like SQLite. No external dependencies. | +| **Performance** | Faster than SQLite for key-value operations, but Jx9 scripting is slow for complex queries. | +| **Query model** | Jx9 scripting language for document queries — but it's interpreted and slow. Manual KV indexes are fast but require same work as LMDB. | +| **Maturity** | Less battle-tested than SQLite or LMDB. Smaller community. | +| **Concurrency** | Single-writer, multiple-reader (like SQLite). | + +**Verdict**: UnQLite is essentially a less mature, less performant LMDB with an optional (slow) document query layer. If you're going embedded KV, LMDB is the better choice. + +### Summary: Document DBs for Nostr Relays + +| Database | Query Fit | C Integration | Performance | Deployment | Verdict | +|----------|-----------|---------------|-------------|------------|---------| +| **CouchDB** | Good | Poor (HTTP) | Slow (HTTP overhead) | Heavy (Erlang) | ❌ Wrong paradigm for embedded relay | +| **MongoDB** | Excellent | Good (libmongoc) | Good | Heavy (server + replica set) | ⚠️ Great queries, bad deployment model | +| **RethinkDB** | Interesting | Poor (no C driver) | Medium | Heavy | ❌ Not production-viable | +| **UnQLite** | Medium | Excellent (single .c) | Medium | None (embedded) | ⚠️ Less mature LMDB alternative | +| **PostgreSQL JSONB** | Excellent | Good (libpq) | Good | Medium (server) | ✅ Best SQL + JSON hybrid | +| **LMDB** | Manual | Excellent (C library) | Best | None (embedded) | ✅ Best raw performance | + +**The key insight**: For a Nostr relay, the "document database" that actually wins is either: +- **PostgreSQL with JSONB** — if you want SQL + JSON indexing + server model +- **LMDB with manual indexes** — if you want maximum performance + embedded model + +The dedicated document databases (CouchDB, MongoDB) are designed for web application backends where HTTP latency is acceptable and operational complexity is managed by a team. For a single-binary C relay, they add overhead without proportional benefit. + +--- + +## The Dashboard Problem: Separate Read Path + +A real-time web dashboard that queries event statistics, subscription analytics, and connection metrics is **exactly the kind of workload that kills a single-process SQLite relay**. Here's why, and how each architecture handles it. + +### Current Architecture: Dashboard Kills the Relay + +Right now, dashboard queries run through the same [`g_db`](src/main.c:49) connection inside the same `lws_service()` event loop. When the dashboard runs analytics from [`event_kinds_view`](src/sql_schema.h:276) or [`time_stats_view`](src/sql_schema.h:297), these are **full table scans** on 65K+ events. Each one takes 100-500ms. While they run, every WebSocket client is frozen. Running these every few seconds for a "real-time" dashboard would make the relay unusable. + +### How Each Architecture Solves This + +```mermaid +flowchart LR + subgraph Current - Everything Shares One Thread + WS1[WebSocket Clients] --> EL[Event Loop] + DASH1[Dashboard Queries] --> EL + EL --> DB1[SQLite g_db] + end +``` + +```mermaid +flowchart LR + subgraph PostgreSQL - True Separation + WS2[WebSocket Clients] --> RELAY[Relay Process] + RELAY -->|connection pool| PG[PostgreSQL] + DASHWEB[Dashboard Web Server] -->|own connection pool| PG + end +``` + +```mermaid +flowchart LR + subgraph LMDB - Multi-Process Reads + WS3[WebSocket Clients] --> RELAY2[Relay Process] + RELAY2 -->|mmap read txn| LMDB1[LMDB File] + DASHWEB2[Dashboard Process] -->|own mmap read txn| LMDB1 + end +``` + +```mermaid +flowchart LR + subgraph SQLite Thread Pool - Partial Separation + WS4[WebSocket Clients] --> EL2[Event Loop] + EL2 -->|dispatch| POOL[Thread Pool] + POOL -->|own connections| DB2[SQLite WAL] + DASH2[Dashboard] -->|also in pool| DB2 + end +``` + +### Detailed Comparison for Dashboard Use Case + +| Factor | SQLite Thread Pool | PostgreSQL | LMDB | +|--------|-------------------|------------|------| +| **Separate dashboard process** | No — SQLite multi-process access is fragile | Yes — any number of processes can connect | Yes — multiple processes can mmap the same file for reads | +| **Dashboard query language** | SQL | SQL | No SQL — must build custom analytics endpoints in C | +| **Real-time refresh cost** | Medium — analytics queries compete with relay queries in the pool | Low — PostgreSQL handles concurrent analytics + relay queries independently | Low for simple metrics, High for complex aggregations | +| **Separate web server** | Difficult — SQLite file locking issues | Easy — any web framework connects to PostgreSQL | Possible — separate process opens LMDB read-only, but must build custom API | +| **Dashboard tech stack** | Must be embedded in relay | **Any**: Grafana, React, Python Flask, Go — all connect to PostgreSQL | Must build custom: LMDB has no standard query interface | + +### This Changes the Recommendation + +The dashboard requirement is a **strong argument for PostgreSQL**: + +1. **True process isolation**: The relay and dashboard are completely separate processes. A heavy dashboard query never affects WebSocket latency. + +2. **Use existing tools**: You could point **Grafana** directly at PostgreSQL and get a beautiful real-time dashboard with zero custom code. Or use any web framework to build a custom one. + +3. **SQL for analytics**: Dashboard queries are inherently analytical — aggregations, time-series, top-N. SQL is the right tool. With LMDB, you'd hand-code every aggregation in C. + +4. **Materialized views**: PostgreSQL can pre-compute expensive analytics: + +```sql +-- Pre-computed, refreshed every 30 seconds in background +CREATE MATERIALIZED VIEW event_stats_mv AS +SELECT kind, COUNT(*) as count, + ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM events), 2) as pct +FROM events GROUP BY kind; + +-- Dashboard reads this instantly, no table scan +SELECT * FROM event_stats_mv ORDER BY count DESC; +``` + +### LMDB + PostgreSQL Hybrid (Best of Both Worlds) + +If you want LMDB's raw performance for the relay hot path AND a rich dashboard: + +```mermaid +flowchart TD + subgraph Relay Process + WS[WebSocket Handler] -->|store and query| LMDB2[LMDB - Events] + WS -->|async write| PG2[PostgreSQL - Analytics] + end + + subgraph Dashboard Process + DASH[Web Dashboard] -->|read only| PG2 + end +``` + +- **LMDB**: Handles all real-time event storage and REQ queries at microsecond latency +- **PostgreSQL**: Receives async copies of events for analytics/dashboard queries +- **Dashboard**: Queries PostgreSQL exclusively, never touches LMDB + +This gives the best of both worlds but adds complexity — two databases to maintain plus async replication logic. + +--- + +## Horizontal Scaling: Multiple Relay Instances in Practice + +With a client-server database, you unlock something fundamentally impossible with SQLite or LMDB: **multiple relay instances sharing the same database**. This section goes deep on how this works in practice. + +### Architecture Overview + +```mermaid +flowchart TD + INTERNET[Internet - Nostr Clients] -->|wss://relay.example.com| NGINX[nginx reverse proxy - TLS termination + load balancing] + + NGINX -->|ws://localhost:8888| R1[c-relay Instance 1 - port 8888] + NGINX -->|ws://localhost:8889| R2[c-relay Instance 2 - port 8889] + NGINX -->|ws://localhost:8890| R3[c-relay Instance 3 - port 8890] + NGINX -->|http://localhost:3000| DASHWEB[Dashboard Web Server] + + R1 -->|libpq connection pool| PGPOOL[PgBouncer - connection pooler] + R2 -->|libpq connection pool| PGPOOL + R3 -->|libpq connection pool| PGPOOL + + PGPOOL -->|pooled connections| PG[PostgreSQL Primary] + DASHWEB -->|read queries| PG + + PG -->|streaming replication| REPLICA[Read Replica - optional] + DASHWEB -.->|heavy analytics| REPLICA +``` + +### How Load Balancing Works in Practice + +nginx already handles TLS termination for most Nostr relays. Adding WebSocket load balancing is a small configuration change: + +```nginx +# nginx.conf - WebSocket load balancing for c-relay +upstream relay_backends { + # ip_hash ensures a client always hits the same instance + # (important for WebSocket session stickiness) + ip_hash; + + server 127.0.0.1:8888; + server 127.0.0.1:8889; + server 127.0.0.1:8890; +} + +server { + listen 443 ssl; + server_name relay.example.com; + + # TLS config... + + location / { + proxy_pass http://relay_backends; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_set_header X-Real-IP $remote_addr; + proxy_read_timeout 86400; # Keep WebSocket alive for 24h + } + + # Dashboard on separate path + location /dashboard { + proxy_pass http://127.0.0.1:3000; + } +} +``` + +**Session stickiness** (`ip_hash`) ensures that once a client connects to Instance 2, all their subsequent WebSocket frames go to Instance 2. This is important because subscriptions are held in-memory per instance. + +### Starting Multiple Instances + +Each instance is the same binary, just on a different port, all pointing to the same PostgreSQL: + +```bash +# Instance 1 +./build/c_relay_x86 --port 8888 --db-host localhost --db-name crelay & + +# Instance 2 +./build/c_relay_x86 --port 8889 --db-host localhost --db-name crelay & + +# Instance 3 +./build/c_relay_x86 --port 8890 --db-host localhost --db-name crelay & +``` + +Or with systemd, you'd use a template unit: + +```ini +# /etc/systemd/system/c-relay@.service +[Unit] +Description=C-Relay Nostr Instance %i +After=postgresql.service + +[Service] +ExecStart=/opt/c-relay/c_relay_x86 --port %i --db-host localhost --db-name crelay +Restart=always +User=c-relay + +[Install] +WantedBy=multi-user.target +``` + +```bash +systemctl enable c-relay@8888 c-relay@8889 c-relay@8890 +systemctl start c-relay@8888 c-relay@8889 c-relay@8890 +``` + +### The Subscription Broadcasting Challenge (Detailed) + +This is the most important technical challenge with multiple instances. When Instance 1 receives a new EVENT and stores it in PostgreSQL, Instance 2 and Instance 3 need to know about it so they can broadcast to their connected subscribers. + +```mermaid +sequenceDiagram + participant Client_A as Client A - connected to Instance 1 + participant I1 as Instance 1 + participant PG as PostgreSQL + participant I2 as Instance 2 + participant I3 as Instance 3 + participant Client_B as Client B - connected to Instance 2 + participant Client_C as Client C - connected to Instance 3 + + Client_A->>I1: EVENT - new kind:1 note + I1->>PG: INSERT INTO events... + PG-->>I1: OK + I1->>I1: broadcast to local subscribers + I1->>PG: NOTIFY new_event with event_id + + Note over PG: PostgreSQL delivers notification to all listeners + + PG-->>I2: NOTIFY: new_event event_id + PG-->>I3: NOTIFY: new_event event_id + + I2->>PG: SELECT event_json FROM events WHERE id = event_id + PG-->>I2: event JSON + I2->>I2: match against local subscriptions + I2->>Client_B: EVENT message - if subscription matches + + I3->>PG: SELECT event_json FROM events WHERE id = event_id + PG-->>I3: event JSON + I3->>I3: match against local subscriptions + I3->>Client_C: EVENT message - if subscription matches +``` + +#### Solution: PostgreSQL LISTEN/NOTIFY + +This is built into PostgreSQL — no additional infrastructure needed: + +```c +// === In the relay's event loop (modified lws_service loop) === + +// Setup: create a dedicated connection for LISTEN +PGconn* notify_conn = PQconnectdb("host=localhost dbname=crelay"); +PQexec(notify_conn, "LISTEN new_event"); +int notify_fd = PQsocket(notify_conn); // Get the socket fd for poll() + +// After storing an event: +void on_event_stored(PGconn* write_conn, const char* event_id) { + char notify_cmd[128]; + snprintf(notify_cmd, sizeof(notify_cmd), + "NOTIFY new_event, '%s'", event_id); + PQexec(write_conn, notify_cmd); +} + +// In the main event loop (runs every lws_service iteration): +void check_cross_instance_events(PGconn* notify_conn) { + // Non-blocking check for notifications + PQconsumeInput(notify_conn); + + PGnotify* notify; + while ((notify = PQnotifies(notify_conn)) != NULL) { + // Another instance stored a new event + const char* event_id = notify->extra; + + // Fetch the event and check against local subscriptions + cJSON* event = db_get_event_by_id(event_id); + if (event) { + broadcast_event_to_subscriptions(event); + cJSON_Delete(event); + } + + PQfreemem(notify); + } +} +``` + +**Performance**: LISTEN/NOTIFY adds ~1-5ms latency for cross-instance delivery. For a Nostr relay, this is imperceptible — clients already expect network latency. + +**Payload limit**: NOTIFY payloads are limited to 8000 bytes. For event IDs (64 hex chars), this is fine. For larger payloads, you'd store the event first and send just the ID. + +#### Alternative: Redis Pub/Sub + +If you later need even lower latency or more sophisticated routing: + +```c +// Using hiredis (Redis C client) +redisContext* redis = redisConnect("127.0.0.1", 6379); + +// After storing event: +redisCommand(redis, "PUBLISH new_event %s", event_json); + +// Subscriber (in each instance): +redisCommand(redis, "SUBSCRIBE new_event"); +// Then poll for messages in the event loop +``` + +Redis adds sub-millisecond pub/sub but requires running a Redis server. For most relays, PostgreSQL LISTEN/NOTIFY is sufficient. + +### Zero-Downtime Deployment in Practice + +```mermaid +sequenceDiagram + participant LB as nginx + participant I1 as Instance 1 - v1.0 + participant I2 as Instance 2 - v1.0 + participant I3 as Instance 3 - v1.0 + participant I1_NEW as Instance 1 - v1.1 + + Note over LB,I3: Normal operation: 3 instances serving traffic + + LB->>I1: Mark upstream as down + Note over I1: Drain: wait for existing connections to close or timeout + I1->>I1: Graceful shutdown + + Note over LB: Traffic now goes to I2 and I3 only + + I1_NEW->>I1_NEW: Start with new binary + I1_NEW->>LB: Health check passes + LB->>I1_NEW: Mark upstream as up + + Note over LB,I1_NEW: Instance 1 now running v1.1, repeat for I2 and I3 +``` + +In practice with nginx: + +```bash +#!/bin/bash +# rolling_deploy.sh - Zero-downtime deployment + +for port in 8888 8889 8890; do + echo "Deploying instance on port $port..." + + # 1. Tell nginx to stop sending new connections + # (mark server as "down" in upstream config, reload nginx) + sed -i "s/server 127.0.0.1:$port;/server 127.0.0.1:$port down;/" /etc/nginx/nginx.conf + nginx -s reload + + # 2. Wait for existing connections to drain (30 seconds) + sleep 30 + + # 3. Stop old instance + systemctl stop c-relay@$port + + # 4. Deploy new binary + cp ./build/c_relay_x86 /opt/c-relay/c_relay_x86 + + # 5. Start new instance + systemctl start c-relay@$port + + # 6. Wait for health check + sleep 5 + + # 7. Re-enable in nginx + sed -i "s/server 127.0.0.1:$port down;/server 127.0.0.1:$port;/" /etc/nginx/nginx.conf + nginx -s reload + + echo "Instance on port $port deployed successfully" +done +``` + +### Connection Pooling with PgBouncer + +Each relay instance needs multiple database connections (for concurrent queries). Without pooling, 3 instances × 10 connections = 30 PostgreSQL backend processes. With PgBouncer: + +```ini +# /etc/pgbouncer/pgbouncer.ini +[databases] +crelay = host=127.0.0.1 port=5432 dbname=crelay + +[pgbouncer] +listen_port = 6432 +listen_addr = 127.0.0.1 +auth_type = md5 +pool_mode = transaction # Return connection to pool after each transaction +max_client_conn = 200 # Total connections from all relay instances +default_pool_size = 20 # Actual PostgreSQL connections +``` + +Relay instances connect to PgBouncer (port 6432) instead of PostgreSQL directly (port 5432). PgBouncer multiplexes 200 client connections onto 20 actual PostgreSQL connections. + +### Scaling Scenarios + +| Scenario | Instances | Connections | Events/hour | Setup | +|----------|-----------|-------------|-------------|-------| +| **Current** | 1 | ~1,200 | 56 | Single process + SQLite | +| **Small upgrade** | 2 | ~2,500 | 500 | 2 instances + PostgreSQL | +| **Medium relay** | 4 | ~5,000 | 5,000 | 4 instances + PostgreSQL + PgBouncer | +| **Large relay** | 8 | ~10,000 | 50,000 | 8 instances + PostgreSQL + read replica | +| **Multi-region** | 2-4 per region | ~50,000 | 500,000 | Multiple servers + PostgreSQL replication | + +### Why This Is Impossible with SQLite/LMDB + +| Feature | SQLite | LMDB | PostgreSQL | MySQL | +|---------|--------|------|------------|-------| +| **Multiple writer processes** | No — file lock | No — single writer | Yes — row-level locking | Yes — row-level locking | +| **Cross-process notifications** | No | No | Yes — LISTEN/NOTIFY | No built-in — need polling or external | +| **Connection pooling** | N/A | N/A | Yes — PgBouncer | Yes — ProxySQL | +| **Read replicas** | No | No | Yes — streaming | Yes — replication | +| **Load balancing** | Impossible | Impossible | Natural fit | Natural fit | +| **Max instances** | 1 | 1 | Unlimited | Unlimited | + +--- + +## SQL Database Comparison: PostgreSQL vs MySQL vs MariaDB + +Since we're considering a client-server SQL database, let's compare the main contenders. + +### PostgreSQL + +**The gold standard for data integrity and advanced features.** + +| Aspect | Details | +|--------|---------| +| **JSONB support** | Native binary JSON with GIN indexing. `SELECT * FROM events WHERE tags @> '[["p","pubkey123"]]'` — indexed, fast. Eliminates the `event_tags` table entirely. | +| **LISTEN/NOTIFY** | Built-in pub/sub for cross-instance event broadcasting. No external dependencies. | +| **Partial indexes** | `CREATE INDEX idx_active ON events(kind) WHERE kind < 20000 OR kind >= 30000` — index only non-ephemeral events, saving space. | +| **MVCC** | True multi-version concurrency. Readers never block writers. No "locked" errors. | +| **Query planner** | Cost-based optimizer with statistics. Automatically chooses the best index for each query. | +| **C client library** | `libpq` — mature, well-documented, supports async queries. | +| **Materialized views** | Pre-compute expensive analytics, refresh on schedule. Dashboard reads are instant. | +| **Full-text search** | Built-in `tsvector` for NIP-50 search support. | +| **Memory** | ~200-500 MB baseline. `shared_buffers` should be ~25% of RAM. | +| **Nostr relay precedent** | Used by **nostream** (TypeScript relay) — proven at scale for Nostr. | + +### MySQL / MariaDB + +**The most widely deployed database. Simpler than PostgreSQL but less feature-rich.** + +| Aspect | Details | +|--------|---------| +| **JSON support** | MySQL 5.7+ has JSON type with `JSON_CONTAINS()` and `JSON_EXTRACT()`. But **no GIN-equivalent index** — JSON queries do full scans or use generated columns with regular indexes. | +| **Cross-instance notifications** | No built-in equivalent to LISTEN/NOTIFY. Would need Redis, polling, or MySQL's binlog streaming. | +| **Partial indexes** | Not supported in MySQL. MariaDB has limited support. | +| **Concurrency** | InnoDB has row-level locking and MVCC, similar to PostgreSQL. | +| **Query planner** | Simpler than PostgreSQL's. Historically weaker for complex queries, but MySQL 8.0 improved significantly. | +| **C client library** | `libmysqlclient` or `libmariadb` — mature, well-documented. | +| **Materialized views** | Not supported natively. Must use tables + triggers or scheduled jobs. | +| **Full-text search** | InnoDB full-text indexes available but less flexible than PostgreSQL's. | +| **Memory** | ~100-300 MB baseline. Generally lighter than PostgreSQL. | +| **Nostr relay precedent** | No major Nostr relay uses MySQL. | + +### MariaDB + +MariaDB is a MySQL fork with some additional features: +- Better JSON support than MySQL (but still no GIN indexes) +- `CONNECT` engine for external data sources +- `ColumnStore` engine for analytics (interesting for dashboard) +- Generally compatible with MySQL client libraries + +### Head-to-Head for Nostr Relay Use Case + +| Feature | PostgreSQL | MySQL 8.0 | MariaDB | +|---------|-----------|-----------|---------| +| **JSON tag indexing** | **GIN index on JSONB — O(log n) lookups** | Generated column + B-tree — works but manual | Similar to MySQL | +| **Eliminates event_tags table** | **Yes — JSONB @> operator with GIN** | No — still need denormalized table or generated columns | No | +| **Cross-instance pub/sub** | **LISTEN/NOTIFY — built-in** | Need external solution (Redis, polling) | Need external solution | +| **Partial indexes** | **Yes — index only what matters** | No | Limited | +| **Materialized views** | **Yes — instant dashboard reads** | No — manual implementation | No | +| **NIP-50 full-text search** | **tsvector — powerful and indexed** | InnoDB FTS — adequate | InnoDB FTS — adequate | +| **Async C client** | **libpq async mode** | libmysqlclient async (MySQL 8.0+) | libmariadb async | +| **Connection pooling** | **PgBouncer — battle-tested** | ProxySQL — good | ProxySQL — good | +| **Ease of setup** | Medium — more config options | Easy — simpler defaults | Easy — simpler defaults | +| **Community/docs** | Excellent | Excellent | Good | +| **Hosting availability** | Every cloud provider | Every cloud provider | Most cloud providers | + +### The Killer Feature: JSONB + GIN Indexes + +This is why PostgreSQL wins for Nostr specifically. Your current schema has the `event_tags` table (4 million rows, ~424 MB data, ~2 GB indexes) solely because SQLite can't efficiently query JSON arrays. + +With PostgreSQL JSONB + GIN: + +```sql +-- PostgreSQL schema (no event_tags table needed!) +CREATE TABLE events ( + id TEXT PRIMARY KEY, + pubkey TEXT NOT NULL, + created_at BIGINT NOT NULL, + kind INTEGER NOT NULL, + content TEXT NOT NULL, + sig TEXT NOT NULL, + tags JSONB NOT NULL DEFAULT '[]', + event_json TEXT NOT NULL +); + +-- GIN index on tags — handles ALL tag queries +CREATE INDEX idx_events_tags ON events USING GIN (tags); + +-- Query: find events with #p tag matching a pubkey +-- This uses the GIN index — O(log n), not a table scan +SELECT event_json FROM events +WHERE tags @> '[["p", "pubkey_hex_here"]]' +AND kind IN (1, 6, 7) +ORDER BY created_at DESC +LIMIT 100; + +-- Query: find events with #e tag +SELECT event_json FROM events +WHERE tags @> '[["e", "event_id_here"]]' +ORDER BY created_at DESC +LIMIT 100; +``` + +**No `event_tags` table. No 4 million denormalized rows. No 2 GB of indexes.** The GIN index on the JSONB `tags` column handles everything in ~100-200 MB. + +Compare to your current SQLite approach in [`handle_req_message()`](src/main.c:1459): +```sql +-- Current: subquery into event_tags table +AND id IN (SELECT event_id FROM event_tags + WHERE tag_name = ? AND tag_value IN (?)) +``` + +The PostgreSQL version is both simpler to write AND faster to execute. + +### MySQL's JSON Workaround + +MySQL can't do GIN indexes on JSON. The workaround is generated columns: + +```sql +-- MySQL: must create generated columns for each tag type you want to index +ALTER TABLE events ADD COLUMN tag_p JSON + GENERATED ALWAYS AS ( + JSON_EXTRACT(tags, '$[*]' ) -- complex extraction needed + ) VIRTUAL; +CREATE INDEX idx_tag_p ON events(tag_p); +``` + +This is fragile, requires a generated column per tag type, and doesn't handle arbitrary tag names like `#g` (geohash), `#t` (hashtag), `#type`, etc. You'd need to know all tag types in advance. + +### Verdict: PostgreSQL Is the Clear Winner for Nostr + +PostgreSQL's JSONB + GIN indexes are **purpose-built** for the exact problem Nostr relays face: querying JSON arrays efficiently. No other SQL database matches this capability. Combined with LISTEN/NOTIFY for multi-instance broadcasting and materialized views for dashboards, PostgreSQL is the optimal SQL database for a Nostr relay. + +MySQL/MariaDB would work, but you'd still need the `event_tags` denormalization table, you'd need Redis or polling for cross-instance events, and you'd need manual materialized view implementations. It's more work for less capability. + +### Cost Perspective + +Running multiple relay instances with PostgreSQL on a single VPS: +- **PostgreSQL**: ~200-500 MB RAM baseline +- **PgBouncer**: ~10 MB RAM +- **Each relay instance**: ~50-100 MB RAM +- **Dashboard web server**: ~50-100 MB RAM +- **4 instances + PostgreSQL + PgBouncer + dashboard**: ~1-2 GB total RAM +- A **$20/month VPS** with 4 cores and 4 GB RAM handles this easily +- A **$40/month VPS** with 8 cores and 8 GB RAM handles 8 instances comfortably + +--- + +## Head-to-Head Comparison for YOUR Workload + +Based on your actual production data (2.7 GB DB, 65K events, 120 REQ/min, 56 writes/hour): + +| Factor | SQLite Thread Pool | PostgreSQL | LMDB | +|--------|-------------------|------------|------| +| **Fixes event loop blocking** | Yes | Yes | Yes — reads so fast they don't block | +| **Fixes 672ms worst-case queries** | Partially — queries still take 672ms but don't block others | Yes — better query planner + GIN indexes | **Yes — sub-millisecond reads** | +| **Fixes 2 GB index bloat** | No — same schema | Yes — more efficient indexes, JSONB eliminates event_tags | **Yes — B+ tree is more compact** | +| **Write throughput** | Adequate for 56 events/hour | Overkill for 56 events/hour | Adequate — single writer like SQLite | +| **Code changes** | ~300-500 lines new code, ~200 lines modified | ~2000-3000 lines rewritten | **~3000-4000 lines rewritten** | +| **Risk** | Low — additive change, SQLite stays | High — complete data layer replacement | **Highest — no SQL, manual indexes** | +| **Deployment change** | None — still single binary | Major — need PostgreSQL server | **None — still embedded library** | +| **Time to implement** | Moderate | Large | **Largest** | +| **Future scaling ceiling** | ~1000 concurrent connections | ~10,000+ concurrent connections | **~10,000+ like strfry** | +| **Rollback difficulty** | Easy — remove thread pool, back to synchronous | Very hard — different database entirely | **Very hard — different paradigm** | +| **Keeps admin SQL API** | Yes | Yes (different SQL dialect) | **No — must rebuild as specific endpoints** | +| **Raw read performance** | Same as current | 5-10x faster | **100-1000x faster** | +| **Operational simplicity** | Same as current | Needs DBA knowledge | **Same as current — single file** | + +--- + +## Recommendation: Phased Approach + +### Phase 1: Database Abstraction Layer (prerequisite for any path) + +Create a `db_ops.h` / `db_ops.c` that wraps all database operations behind a clean interface: + +```c +// db_ops.h - Database operations interface +typedef struct db_result db_result_t; +typedef struct db_event_filter db_event_filter_t; + +// Core event operations +int db_store_event(const char* event_json, const char* id, const char* pubkey, + int kind, long created_at, const char* tags_json); +int db_event_exists(const char* event_id); +int db_delete_event(const char* event_id); +const char* db_get_event_json(const char* event_id); // returns pointer, caller must not free for LMDB + +// Query operations - backend-agnostic filter +db_result_t* db_query_events(const db_event_filter_t* filter); +int db_count_events(const db_event_filter_t* filter); +const char* db_result_next(db_result_t* result); // returns event_json pointer +void db_result_free(db_result_t* result); + +// Config operations +const char* db_get_config(const char* key); +int db_set_config(const char* key, const char* value); + +// Lifecycle +int db_init(const char* path); +void db_close(void); +``` + +This consolidates the 258 scattered `sqlite3_*` calls into one file. The interface is designed to work with **any** backend: +- SQLite: `db_result_next()` copies from `sqlite3_column_text()` +- LMDB: `db_result_next()` returns a zero-copy pointer into mmap'd memory +- PostgreSQL: `db_result_next()` returns from `PQgetvalue()` + +### Phase 2: Choose Your Backend + +#### Path A: Thread Pool (fastest to implement, lowest risk) +- Keep SQLite, add worker threads for reads +- **Best if**: You want immediate relief and minimal disruption +- **Gets you**: Event loop unblocked, 4-8x effective throughput + +#### Path B: LMDB (maximum performance, proven for Nostr) +- Replace SQLite with LMDB behind the abstraction layer +- **Best if**: You want strfry-level performance and are willing to invest in the rewrite +- **Gets you**: Sub-millisecond reads, zero-copy, 100-1000x faster queries, compact storage +- **Loses**: Admin SQL query API, ad-hoc analytics queries + +#### Path C: PostgreSQL (maximum flexibility, operational overhead) +- Replace SQLite with PostgreSQL behind the abstraction layer +- **Best if**: You want SQL + concurrency + operational tooling +- **Gets you**: True concurrent writes, superior query planner, GIN indexes, replication +- **Loses**: Embedded simplicity, single-binary deployment + +### Phase 3 (if LMDB chosen): Rebuild Admin Analytics + +Replace the SQL-based admin API with purpose-built LMDB query endpoints: +- Event kind distribution → iterate events db, aggregate in C +- Top pubkeys → maintain a separate counter database +- Time-based stats → range scan on time index +- Or: keep a small SQLite database alongside LMDB just for analytics/config (hybrid approach) + +--- + +## Bottom Line + +Here's how I'd rank the options for **your specific situation** (65K events, 2.7 GB DB, 120 REQ/min, medium traffic relay): + +### If you want the fastest fix with lowest risk: +**→ Thread Pool (Path A)**. Unblocks the event loop immediately. Your 672ms queries still take 672ms, but they no longer freeze every other client. This buys you time to plan a bigger change. + +### If you want the best long-term architecture: +**→ LMDB (Path B)**. This is what the fastest Nostr relay in existence uses, and for good reason. Your 672ms queries become sub-millisecond. Your 2.7 GB database becomes ~400 MB. You don't need a thread pool because reads are so fast they can run inline. The downside is it's the biggest rewrite and you lose SQL flexibility. + +### If you want SQL + performance: +**→ PostgreSQL (Path C)**. You keep SQL, get concurrent writes, get a better query planner, and get operational tooling. But you add an external dependency and deployment complexity. + +### The pragmatic path: +**Phase 1 (abstraction layer) → Phase 2A (thread pool for immediate relief) → Phase 2B (LMDB migration behind the abstraction layer)**. This gives you immediate improvement while building toward the optimal architecture. The abstraction layer from Phase 1 makes the LMDB migration a contained change rather than a scattered rewrite across 8 files. + +**Regardless of which backend you choose, Phase 1 (abstraction layer) is the right first step.** It cleans up the codebase, makes testing easier, and enables any future backend swap. diff --git a/src/db_ops.c b/src/db_ops.c new file mode 100644 index 0000000..b286782 --- /dev/null +++ b/src/db_ops.c @@ -0,0 +1,292 @@ +#define _GNU_SOURCE + +#include "db_ops.h" +#include "debug.h" +#include "config.h" + +#include +#include + +// Database handle owned by main.c today. +extern sqlite3* g_db; +extern char g_database_path[512]; + +int db_is_available(void) { + return g_db != NULL; +} + +sqlite3* db_get_handle(void) { + return g_db; +} + +int db_log_subscription_created(const char* sub_id, const char* wsi_ptr, + const char* client_ip, const char* filter_json) { + if (!g_db || !sub_id || !wsi_ptr || !client_ip) return -1; + + const char* sql = + "INSERT OR REPLACE INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type, filter_json) " + "VALUES (?, ?, ?, 'created', ?)"; + + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + sqlite3_bind_text(stmt, 1, sub_id, -1, SQLITE_TRANSIENT); + sqlite3_bind_text(stmt, 2, wsi_ptr, -1, SQLITE_TRANSIENT); + sqlite3_bind_text(stmt, 3, client_ip, -1, SQLITE_TRANSIENT); + sqlite3_bind_text(stmt, 4, filter_json ? filter_json : "[]", -1, SQLITE_TRANSIENT); + + int rc = sqlite3_step(stmt); + sqlite3_finalize(stmt); + return (rc == SQLITE_DONE) ? 0 : -1; +} + +int db_log_subscription_closed(const char* sub_id, const char* client_ip) { + if (!g_db || !sub_id) return -1; + + const char* insert_sql = + "INSERT INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type) " + "VALUES (?, '', ?, 'closed')"; + + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(g_db, insert_sql, -1, &stmt, NULL) == SQLITE_OK) { + sqlite3_bind_text(stmt, 1, sub_id, -1, SQLITE_TRANSIENT); + sqlite3_bind_text(stmt, 2, client_ip ? client_ip : "unknown", -1, SQLITE_TRANSIENT); + sqlite3_step(stmt); + sqlite3_finalize(stmt); + } + + const char* update_sql = + "UPDATE subscriptions " + "SET ended_at = strftime('%s', 'now') " + "WHERE subscription_id = ? AND event_type = 'created' AND ended_at IS NULL"; + + if (sqlite3_prepare_v2(g_db, update_sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + sqlite3_bind_text(stmt, 1, sub_id, -1, SQLITE_TRANSIENT); + int rc = sqlite3_step(stmt); + sqlite3_finalize(stmt); + + return (rc == SQLITE_DONE) ? 0 : -1; +} + +int db_log_subscription_disconnected(const char* client_ip) { + if (!g_db || !client_ip) return -1; + + const char* update_sql = + "UPDATE subscriptions " + "SET ended_at = strftime('%s', 'now') " + "WHERE client_ip = ? AND event_type = 'created' AND ended_at IS NULL"; + + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(g_db, update_sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + sqlite3_bind_text(stmt, 1, client_ip, -1, SQLITE_TRANSIENT); + int rc = sqlite3_step(stmt); + int changes = sqlite3_changes(g_db); + sqlite3_finalize(stmt); + if (rc != SQLITE_DONE) return -1; + + if (changes > 0) { + const char* insert_sql = + "INSERT INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type) " + "VALUES ('disconnect', '', ?, 'disconnected')"; + + if (sqlite3_prepare_v2(g_db, insert_sql, -1, &stmt, NULL) == SQLITE_OK) { + sqlite3_bind_text(stmt, 1, client_ip, -1, SQLITE_TRANSIENT); + sqlite3_step(stmt); + sqlite3_finalize(stmt); + } + } + + return changes; +} + +int db_update_subscription_events_sent(const char* sub_id, int events_sent) { + if (!g_db || !sub_id) return -1; + + const char* sql = + "UPDATE subscriptions " + "SET events_sent = ? " + "WHERE subscription_id = ? AND event_type = 'created'"; + + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + sqlite3_bind_int(stmt, 1, events_sent); + sqlite3_bind_text(stmt, 2, sub_id, -1, SQLITE_TRANSIENT); + + int rc = sqlite3_step(stmt); + sqlite3_finalize(stmt); + return (rc == SQLITE_DONE) ? 0 : -1; +} + +int db_cleanup_orphaned_subscriptions(void) { + if (!g_db) return -1; + + const char* sql = + "UPDATE subscriptions " + "SET ended_at = strftime('%s', 'now') " + "WHERE event_type = 'created' AND ended_at IS NULL"; + + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + int rc = sqlite3_step(stmt); + int changes = sqlite3_changes(g_db); + sqlite3_finalize(stmt); + + return (rc == SQLITE_DONE) ? changes : -1; +} + +int db_get_event_pubkey(const char* event_id, char* pubkey_out, size_t pubkey_out_size) { + if (!g_db || !event_id || !pubkey_out || pubkey_out_size == 0) return -1; + + const char* sql = "SELECT pubkey FROM events WHERE id = ?"; + sqlite3_stmt* stmt = NULL; + + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + sqlite3_bind_text(stmt, 1, event_id, -1, SQLITE_TRANSIENT); + int rc = sqlite3_step(stmt); + if (rc == SQLITE_ROW) { + const char* pubkey = (const char*)sqlite3_column_text(stmt, 0); + if (pubkey) { + snprintf(pubkey_out, pubkey_out_size, "%s", pubkey); + sqlite3_finalize(stmt); + return 1; + } + } + + sqlite3_finalize(stmt); + return 0; +} + +int db_delete_event_by_id(const char* event_id, const char* requester_pubkey) { + if (!g_db || !event_id || !requester_pubkey) return -1; + + const char* sql = "DELETE FROM events WHERE id = ? AND pubkey = ?"; + sqlite3_stmt* stmt = NULL; + + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + + sqlite3_bind_text(stmt, 1, event_id, -1, SQLITE_TRANSIENT); + sqlite3_bind_text(stmt, 2, requester_pubkey, -1, SQLITE_TRANSIENT); + + int rc = sqlite3_step(stmt); + int changes = sqlite3_changes(g_db); + sqlite3_finalize(stmt); + + if (rc != SQLITE_DONE) return -1; + return changes; +} + +int db_delete_events_by_address(const char* pubkey, int kind, + const char* d_tag, long before_timestamp) { + if (!g_db || !pubkey) return -1; + + const char* sql_with_d = + "DELETE FROM events WHERE kind = ? AND pubkey = ? AND created_at <= ? " + "AND json_extract(tags, '$[*]') LIKE '%[\"d\",\"' || ? || '\"]%'"; + const char* sql_no_d = + "DELETE FROM events WHERE kind = ? AND pubkey = ? AND created_at <= ?"; + + const int has_d = (d_tag && d_tag[0] != '\0'); + sqlite3_stmt* stmt = NULL; + + if (sqlite3_prepare_v2(g_db, has_d ? sql_with_d : sql_no_d, -1, &stmt, NULL) != SQLITE_OK) { + return -1; + } + + sqlite3_bind_int(stmt, 1, kind); + sqlite3_bind_text(stmt, 2, pubkey, -1, SQLITE_TRANSIENT); + sqlite3_bind_int64(stmt, 3, before_timestamp); + if (has_d) { + sqlite3_bind_text(stmt, 4, d_tag, -1, SQLITE_TRANSIENT); + } + + int rc = sqlite3_step(stmt); + int changes = sqlite3_changes(g_db); + sqlite3_finalize(stmt); + + if (rc != SQLITE_DONE) return -1; + return changes; +} + +static int db_scalar_exists_open(sqlite3* db, const char* sql, const char* value) { + sqlite3_stmt* stmt = NULL; + if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0; + + if (value) sqlite3_bind_text(stmt, 1, value, -1, SQLITE_TRANSIENT); + + int exists = (sqlite3_step(stmt) == SQLITE_ROW) ? 1 : 0; + sqlite3_finalize(stmt); + return exists; +} + +int db_is_pubkey_blacklisted(const char* pubkey) { + if (!pubkey || g_database_path[0] == '\0') return 0; + + sqlite3* db = NULL; + if (sqlite3_open_v2(g_database_path, &db, SQLITE_OPEN_READONLY, NULL) != SQLITE_OK) return 0; + + const char* sql = + "SELECT 1 FROM auth_rules WHERE rule_type = 'blacklist' " + "AND pattern_type = 'pubkey' AND pattern_value = ? AND active = 1 LIMIT 1"; + + int exists = db_scalar_exists_open(db, sql, pubkey); + sqlite3_close(db); + return exists; +} + +int db_is_hash_blacklisted(const char* resource_hash) { + if (!resource_hash || g_database_path[0] == '\0') return 0; + + sqlite3* db = NULL; + if (sqlite3_open_v2(g_database_path, &db, SQLITE_OPEN_READONLY, NULL) != SQLITE_OK) return 0; + + const char* sql = + "SELECT 1 FROM auth_rules WHERE rule_type = 'blacklist' " + "AND pattern_type = 'hash' AND pattern_value = ? AND active = 1 LIMIT 1"; + + int exists = db_scalar_exists_open(db, sql, resource_hash); + sqlite3_close(db); + return exists; +} + +int db_is_pubkey_whitelisted(const char* pubkey) { + if (!pubkey || g_database_path[0] == '\0') return 0; + + sqlite3* db = NULL; + if (sqlite3_open_v2(g_database_path, &db, SQLITE_OPEN_READONLY, NULL) != SQLITE_OK) return 0; + + const char* sql = + "SELECT 1 FROM auth_rules WHERE rule_type IN ('whitelist', 'wot_whitelist') " + "AND pattern_type = 'pubkey' AND pattern_value = ? AND active = 1 LIMIT 1"; + + int exists = db_scalar_exists_open(db, sql, pubkey); + sqlite3_close(db); + return exists; +} + +int db_count_active_whitelist_rules(void) { + if (g_database_path[0] == '\0') return 0; + + sqlite3* db = NULL; + sqlite3_stmt* stmt = NULL; + int count = 0; + + if (sqlite3_open_v2(g_database_path, &db, SQLITE_OPEN_READONLY, NULL) != SQLITE_OK) return 0; + + const char* sql = + "SELECT COUNT(*) FROM auth_rules WHERE rule_type IN ('whitelist', 'wot_whitelist') " + "AND pattern_type = 'pubkey' AND active = 1 LIMIT 1"; + + if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) == SQLITE_OK) { + if (sqlite3_step(stmt) == SQLITE_ROW) { + count = sqlite3_column_int(stmt, 0); + } + sqlite3_finalize(stmt); + } + + sqlite3_close(db); + return count; +} diff --git a/src/db_ops.h b/src/db_ops.h new file mode 100644 index 0000000..e7b8e0e --- /dev/null +++ b/src/db_ops.h @@ -0,0 +1,31 @@ +#ifndef DB_OPS_H +#define DB_OPS_H + +#include +#include + +// Generic helpers +int db_is_available(void); +sqlite3* db_get_handle(void); + +// Subscription logging +int db_log_subscription_created(const char* sub_id, const char* wsi_ptr, + const char* client_ip, const char* filter_json); +int db_log_subscription_closed(const char* sub_id, const char* client_ip); +int db_log_subscription_disconnected(const char* client_ip); +int db_update_subscription_events_sent(const char* sub_id, int events_sent); +int db_cleanup_orphaned_subscriptions(void); + +// NIP-09 event deletion helpers +int db_get_event_pubkey(const char* event_id, char* pubkey_out, size_t pubkey_out_size); +int db_delete_event_by_id(const char* event_id, const char* requester_pubkey); +int db_delete_events_by_address(const char* pubkey, int kind, + const char* d_tag, long before_timestamp); + +// Auth rule checks for request validator +int db_is_pubkey_blacklisted(const char* pubkey); +int db_is_hash_blacklisted(const char* resource_hash); +int db_is_pubkey_whitelisted(const char* pubkey); +int db_count_active_whitelist_rules(void); + +#endif // DB_OPS_H diff --git a/src/ip_ban.c b/src/ip_ban.c index dc608a2..6a84b73 100644 --- a/src/ip_ban.c +++ b/src/ip_ban.c @@ -2,6 +2,7 @@ #include "ip_ban.h" #include "debug.h" #include "config.h" +#include "db_ops.h" #include #include #include @@ -100,7 +101,8 @@ void ip_ban_init(void) { DEBUG_LOG("IP ban table initialized (%d slots)", IP_BAN_TABLE_SIZE); } -void ip_ban_load_from_db(sqlite3* db) { +void ip_ban_load_from_db(void) { + sqlite3* db = db_get_handle(); if (!db || !g_initialized) return; // Create table if it doesn't exist (handles existing databases) @@ -193,7 +195,8 @@ void ip_ban_load_from_db(sqlite3* db) { DEBUG_WARN("IP ban table loaded: %d IPs restored (%d still banned)", loaded, still_banned); } -void ip_ban_save_to_db(sqlite3* db) { +void ip_ban_save_to_db(void) { + sqlite3* db = db_get_handle(); if (!db || !g_initialized) return; const char* upsert_sql = @@ -547,7 +550,7 @@ int ip_ban_get_tracked_count(void) { return count; } -void ip_ban_log_stats(sqlite3* db) { +void ip_ban_log_stats(void) { if (!g_initialized) return; static time_t last_log = 0; @@ -557,9 +560,7 @@ void ip_ban_log_stats(sqlite3* db) { last_log = now; // Save to DB every 5 minutes - if (db) { - ip_ban_save_to_db(db); - } + ip_ban_save_to_db(); pthread_mutex_lock(&g_ban_mutex); int auth_banned_count = 0; diff --git a/src/ip_ban.h b/src/ip_ban.h index 87ce31d..ff44fda 100644 --- a/src/ip_ban.h +++ b/src/ip_ban.h @@ -15,7 +15,6 @@ // auth_fail_ban_duration_sec int default: 300 (initial ban duration, doubles each time) #include -#include // Maximum number of IPs tracked simultaneously (fixed-size, no malloc) #define IP_BAN_TABLE_SIZE 4096 @@ -25,11 +24,11 @@ void ip_ban_init(void); // Load ban state from the ip_bans database table. // Call after ip_ban_init() and after the database is open. -void ip_ban_load_from_db(sqlite3* db); +void ip_ban_load_from_db(void); // Save current ban state to the ip_bans database table. // Called every 5 minutes from the maintenance timer. -void ip_ban_save_to_db(sqlite3* db); +void ip_ban_save_to_db(void); // Check if an IP is currently banned. // Returns 1 if banned (connection should be rejected), 0 if allowed. @@ -57,7 +56,7 @@ void ip_ban_cleanup(void); // Emit a periodic WARN-level log summary of banned IPs (every 5 minutes). // Also saves state to DB. -void ip_ban_log_stats(sqlite3* db); +void ip_ban_log_stats(void); // Get stats for logging/monitoring int ip_ban_get_banned_count(void); diff --git a/src/main.c b/src/main.c index f74059a..28b8bb7 100644 --- a/src/main.c +++ b/src/main.c @@ -26,6 +26,7 @@ #include "websockets.h" // WebSocket protocol implementation #include "subscriptions.h" // Subscription management system #include "debug.h" // Debug system +#include "thread_pool.h" // Thread pool scaffold // Forward declarations for unified request validator int nostr_validate_unified_request(const char* json_string, size_t json_length); @@ -162,9 +163,12 @@ int handle_nip11_http_request(struct lws* wsi, const char* accept_header); // Forward declaration for WebSocket relay server int start_websocket_relay(int port_override, int strict_port); +// Thread pool wake callback (called by worker threads) +static void wake_event_loop_from_thread_pool(void* ctx); + // Forward declarations for IP ban system void ip_ban_init(void); -void ip_ban_load_from_db(sqlite3* db); +void ip_ban_load_from_db(void); // Forward declarations for NIP-13 PoW handling (now in nip013.c) @@ -252,6 +256,13 @@ void signal_handler(int sig) { } } +static void wake_event_loop_from_thread_pool(void* ctx) { + (void)ctx; + if (ws_context) { + lws_cancel_service(ws_context); + } +} + ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // NOTICE MESSAGE SUPPORT @@ -785,6 +796,10 @@ int store_event_tags(const char* event_id, cJSON* tags) { // Store event in database int store_event(cJSON* event) { + if (thread_pool_is_running()) { + DEBUG_TRACE("Thread pool scaffold active: store_event() still using synchronous DB path"); + } + if (!g_db || !event) { return -1; } @@ -1099,6 +1114,10 @@ static int is_only_kind_99999_request(cJSON* filters) { } int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { + if (thread_pool_is_running()) { + DEBUG_TRACE("Thread pool scaffold active: handle_req_message() still using synchronous DB path"); + } + if (!cJSON_IsArray(filters)) { DEBUG_ERROR("REQ filters is not an array"); return 0; @@ -2251,11 +2270,35 @@ int main(int argc, char* argv[]) { // Initialize IP ban table and load persisted state from database ip_ban_init(); - ip_ban_load_from_db(g_db); + ip_ban_load_from_db(); + + // Optional thread pool scaffold initialization (execution wiring is future work) + int thread_pool_enabled = get_config_bool("thread_pool_enabled", 0); + int thread_pool_initialized = 0; + if (thread_pool_enabled) { + thread_pool_config_t tp_cfg; + memset(&tp_cfg, 0, sizeof(tp_cfg)); + tp_cfg.reader_threads = get_config_int("thread_pool_readers", 4); + tp_cfg.max_queue_depth = get_config_int("thread_pool_max_queue_depth", 4096); + tp_cfg.db_path = g_database_path; + tp_cfg.wake_loop_cb = wake_event_loop_from_thread_pool; + tp_cfg.wake_loop_ctx = NULL; + + if (thread_pool_init(&tp_cfg) == 0) { + thread_pool_initialized = 1; + DEBUG_INFO("Thread pool scaffold enabled (%d readers)", tp_cfg.reader_threads); + } else { + DEBUG_WARN("Failed to initialize thread pool scaffold; continuing in synchronous mode"); + } + } // Start WebSocket Nostr relay server (port from CLI override or configuration) int result = start_websocket_relay(cli_options.port_override, cli_options.strict_port); // Use CLI port override if specified, otherwise config - + + if (thread_pool_initialized) { + thread_pool_shutdown(); + } + // Cleanup cleanup_relay_info(); ginxsom_request_validator_cleanup(); diff --git a/src/main.h b/src/main.h index 308e9f5..75e4bc1 100644 --- a/src/main.h +++ b/src/main.h @@ -11,10 +11,10 @@ // Version information (auto-updated by build system) // Using CRELAY_ prefix to avoid conflicts with nostr_core_lib VERSION macros -#define CRELAY_VERSION_MAJOR 1 -#define CRELAY_VERSION_MINOR 2 -#define CRELAY_VERSION_PATCH 56 -#define CRELAY_VERSION "v1.2.56" +#define CRELAY_VERSION_MAJOR 2 +#define CRELAY_VERSION_MINOR 0 +#define CRELAY_VERSION_PATCH 0 +#define CRELAY_VERSION "v2.0.0" // Relay metadata (authoritative source for NIP-11 information) #define RELAY_NAME "C-Relay" diff --git a/src/nip009.c b/src/nip009.c index 020ff96..77e9258 100644 --- a/src/nip009.c +++ b/src/nip009.c @@ -7,7 +7,7 @@ ///////////////////////////////////////////////////////////////////////////////////////// #include #include "debug.h" -#include +#include "db_ops.h" #include #include #include @@ -21,10 +21,6 @@ int store_event(cJSON* event); int delete_events_by_id(const char* requester_pubkey, cJSON* event_ids); int delete_events_by_address(const char* requester_pubkey, cJSON* addresses, long deletion_timestamp); -// Global database variable -extern sqlite3* g_db; - - // Handle NIP-09 deletion request event (kind 5) int handle_deletion_request(cJSON* event, char* error_message, size_t error_size) { if (!event) { @@ -146,97 +142,72 @@ int handle_deletion_request(cJSON* event, char* error_message, size_t error_size // Delete events by ID (with pubkey authorization) int delete_events_by_id(const char* requester_pubkey, cJSON* event_ids) { - if (!g_db || !requester_pubkey || !event_ids || !cJSON_IsArray(event_ids)) { + if (!db_is_available() || !requester_pubkey || !event_ids || !cJSON_IsArray(event_ids)) { return 0; } - + int deleted_count = 0; - + cJSON* event_id = NULL; cJSON_ArrayForEach(event_id, event_ids) { if (!cJSON_IsString(event_id)) { continue; } - + const char* id = cJSON_GetStringValue(event_id); - - // First check if event exists and if requester is authorized - const char* check_sql = "SELECT pubkey FROM events WHERE id = ?"; - sqlite3_stmt* check_stmt; - - int rc = sqlite3_prepare_v2(g_db, check_sql, -1, &check_stmt, NULL); - if (rc != SQLITE_OK) { + char event_pubkey[65] = {0}; + int exists = db_get_event_pubkey(id, event_pubkey, sizeof(event_pubkey)); + if (exists <= 0) { continue; } - - sqlite3_bind_text(check_stmt, 1, id, -1, SQLITE_STATIC); - - if (sqlite3_step(check_stmt) == SQLITE_ROW) { - const char* event_pubkey = (char*)sqlite3_column_text(check_stmt, 0); - - // Only delete if the requester is the author - if (event_pubkey && strcmp(event_pubkey, requester_pubkey) == 0) { - sqlite3_finalize(check_stmt); - - // Delete the event - const char* delete_sql = "DELETE FROM events WHERE id = ? AND pubkey = ?"; - sqlite3_stmt* delete_stmt; - - rc = sqlite3_prepare_v2(g_db, delete_sql, -1, &delete_stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(delete_stmt, 1, id, -1, SQLITE_STATIC); - sqlite3_bind_text(delete_stmt, 2, requester_pubkey, -1, SQLITE_STATIC); - - if (sqlite3_step(delete_stmt) == SQLITE_DONE && sqlite3_changes(g_db) > 0) { - deleted_count++; - } - sqlite3_finalize(delete_stmt); - } - } else { - sqlite3_finalize(check_stmt); - char warning_msg[128]; - snprintf(warning_msg, sizeof(warning_msg), "Unauthorized deletion attempt for event: %.16s...", id); - DEBUG_WARN(warning_msg); + + // Only delete if the requester is the author + if (strcmp(event_pubkey, requester_pubkey) == 0) { + int changes = db_delete_event_by_id(id, requester_pubkey); + if (changes > 0) { + deleted_count += changes; } } else { - sqlite3_finalize(check_stmt); + char warning_msg[128]; + snprintf(warning_msg, sizeof(warning_msg), "Unauthorized deletion attempt for event: %.16s...", id); + DEBUG_WARN(warning_msg); } } - + return deleted_count; } // Delete events by addressable reference (kind:pubkey:d-identifier) int delete_events_by_address(const char* requester_pubkey, cJSON* addresses, long deletion_timestamp) { - if (!g_db || !requester_pubkey || !addresses || !cJSON_IsArray(addresses)) { + if (!db_is_available() || !requester_pubkey || !addresses || !cJSON_IsArray(addresses)) { return 0; } - + int deleted_count = 0; - + cJSON* address = NULL; cJSON_ArrayForEach(address, addresses) { if (!cJSON_IsString(address)) { continue; } - + const char* addr = cJSON_GetStringValue(address); - + // Parse address format: kind:pubkey:d-identifier char* addr_copy = strdup(addr); if (!addr_copy) continue; - + char* kind_str = strtok(addr_copy, ":"); char* pubkey_str = strtok(NULL, ":"); char* d_identifier = strtok(NULL, ":"); - + if (!kind_str || !pubkey_str) { free(addr_copy); continue; } - + int kind = atoi(kind_str); - + // Only delete if the requester is the author if (strcmp(pubkey_str, requester_pubkey) != 0) { free(addr_copy); @@ -245,42 +216,15 @@ int delete_events_by_address(const char* requester_pubkey, cJSON* addresses, lon DEBUG_WARN(warning_msg); continue; } - - // Build deletion query based on whether we have d-identifier - const char* delete_sql; - sqlite3_stmt* delete_stmt; - - if (d_identifier && strlen(d_identifier) > 0) { - // Delete specific addressable event with d-tag - delete_sql = "DELETE FROM events WHERE kind = ? AND pubkey = ? AND created_at <= ? " - "AND json_extract(tags, '$[*]') LIKE '%[\"d\",\"' || ? || '\"]%'"; - } else { - // Delete all events of this kind by this author up to deletion timestamp - delete_sql = "DELETE FROM events WHERE kind = ? AND pubkey = ? AND created_at <= ?"; + + int changes = db_delete_events_by_address(requester_pubkey, kind, d_identifier, deletion_timestamp); + if (changes > 0) { + deleted_count += changes; } - - int rc = sqlite3_prepare_v2(g_db, delete_sql, -1, &delete_stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_int(delete_stmt, 1, kind); - sqlite3_bind_text(delete_stmt, 2, requester_pubkey, -1, SQLITE_STATIC); - sqlite3_bind_int64(delete_stmt, 3, deletion_timestamp); - - if (d_identifier && strlen(d_identifier) > 0) { - sqlite3_bind_text(delete_stmt, 4, d_identifier, -1, SQLITE_STATIC); - } - - if (sqlite3_step(delete_stmt) == SQLITE_DONE) { - int changes = sqlite3_changes(g_db); - if (changes > 0) { - deleted_count += changes; - } - } - sqlite3_finalize(delete_stmt); - } - + free(addr_copy); } - + return deleted_count; } diff --git a/src/request_validator.c b/src/request_validator.c index 439525c..af2ef05 100644 --- a/src/request_validator.c +++ b/src/request_validator.c @@ -17,17 +17,13 @@ #include "../nostr_core_lib/nostr_core/utils.h" #include "debug.h" // C-relay debug system #include "config.h" // C-relay configuration system -#include +#include "db_ops.h" #include #include #include #include #include -// External references to C-relay global state -extern sqlite3* g_db; -extern char g_database_path[512]; - // Forward declaration for C-relay event storage function extern int store_event(cJSON* event); @@ -530,112 +526,40 @@ static int reload_auth_config(void) { */ int check_database_auth_rules(const char *pubkey, const char *operation __attribute__((unused)), const char *resource_hash) { - sqlite3 *db = NULL; - sqlite3_stmt *stmt = NULL; - int rc; - DEBUG_TRACE("Checking auth rules for pubkey: %s", pubkey); if (!pubkey) { return NOSTR_ERROR_INVALID_INPUT; } - // Open database using global database path - if (strlen(g_database_path) == 0) { - return NOSTR_SUCCESS; // Default allow on DB error - } - - rc = sqlite3_open_v2(g_database_path, &db, SQLITE_OPEN_READONLY, NULL); - if (rc != SQLITE_OK) { - return NOSTR_SUCCESS; // Default allow on DB error - } - // Step 1: Check pubkey blacklist (highest priority) - const char *blacklist_sql = - "SELECT rule_type FROM auth_rules WHERE rule_type = " - "'blacklist' AND pattern_type = 'pubkey' AND pattern_value = ? AND active = 1 LIMIT 1"; - DEBUG_TRACE("Blacklist SQL: %s", blacklist_sql); - rc = sqlite3_prepare_v2(db, blacklist_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, pubkey, -1, SQLITE_STATIC); - - int step_result = sqlite3_step(stmt); - DEBUG_TRACE("Blacklist query result: %s", step_result == SQLITE_ROW ? "FOUND" : "NOT_FOUND"); - - if (step_result == SQLITE_ROW) { - DEBUG_TRACE("BLACKLIST HIT: Denying access for pubkey: %s", pubkey); - // Set specific violation details for status code mapping - strcpy(g_last_rule_violation.violation_type, "pubkey_blacklist"); - sprintf(g_last_rule_violation.reason, "Public key blacklisted"); - - sqlite3_finalize(stmt); - sqlite3_close(db); - return NOSTR_ERROR_AUTH_REQUIRED; - } - sqlite3_finalize(stmt); + if (db_is_pubkey_blacklisted(pubkey)) { + DEBUG_TRACE("BLACKLIST HIT: Denying access for pubkey: %s", pubkey); + strcpy(g_last_rule_violation.violation_type, "pubkey_blacklist"); + sprintf(g_last_rule_violation.reason, "Public key blacklisted"); + return NOSTR_ERROR_AUTH_REQUIRED; } // Step 2: Check hash blacklist - if (resource_hash) { - const char *hash_blacklist_sql = - "SELECT rule_type FROM auth_rules WHERE rule_type = " - "'blacklist' AND pattern_type = 'hash' AND pattern_value = ? AND active = 1 LIMIT 1"; - rc = sqlite3_prepare_v2(db, hash_blacklist_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, resource_hash, -1, SQLITE_STATIC); - - if (sqlite3_step(stmt) == SQLITE_ROW) { - // Set specific violation details for status code mapping - strcpy(g_last_rule_violation.violation_type, "hash_blacklist"); - sprintf(g_last_rule_violation.reason, "File hash blacklisted"); - - sqlite3_finalize(stmt); - sqlite3_close(db); - return NOSTR_ERROR_AUTH_REQUIRED; - } - sqlite3_finalize(stmt); - } + if (resource_hash && db_is_hash_blacklisted(resource_hash)) { + strcpy(g_last_rule_violation.violation_type, "hash_blacklist"); + sprintf(g_last_rule_violation.reason, "File hash blacklisted"); + return NOSTR_ERROR_AUTH_REQUIRED; } // Step 3: Check pubkey whitelist (includes wot_whitelist) - const char *whitelist_sql = - "SELECT rule_type FROM auth_rules WHERE rule_type IN " - "('whitelist', 'wot_whitelist') AND pattern_type = 'pubkey' AND pattern_value = ? AND active = 1 LIMIT 1"; - rc = sqlite3_prepare_v2(db, whitelist_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, pubkey, -1, SQLITE_STATIC); - - if (sqlite3_step(stmt) == SQLITE_ROW) { - sqlite3_finalize(stmt); - sqlite3_close(db); - return NOSTR_SUCCESS; // Allow whitelisted pubkey - } - sqlite3_finalize(stmt); + if (db_is_pubkey_whitelisted(pubkey)) { + return NOSTR_SUCCESS; // Allow whitelisted pubkey } - // Step 4: Check if any whitelist rules exist - if yes, deny by default (includes wot_whitelist) - const char *whitelist_exists_sql = - "SELECT COUNT(*) FROM auth_rules WHERE rule_type IN ('whitelist', 'wot_whitelist') " - "AND pattern_type = 'pubkey' AND active = 1 LIMIT 1"; - rc = sqlite3_prepare_v2(db, whitelist_exists_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - if (sqlite3_step(stmt) == SQLITE_ROW) { - int whitelist_count = sqlite3_column_int(stmt, 0); - if (whitelist_count > 0) { - // Set specific violation details for status code mapping - strcpy(g_last_rule_violation.violation_type, "whitelist_violation"); - strcpy(g_last_rule_violation.reason, - "Public key not whitelisted for this operation"); - - sqlite3_finalize(stmt); - sqlite3_close(db); - return NOSTR_ERROR_AUTH_REQUIRED; - } - } - sqlite3_finalize(stmt); + // Step 4: If any whitelist rules exist, deny by default + if (db_count_active_whitelist_rules() > 0) { + strcpy(g_last_rule_violation.violation_type, "whitelist_violation"); + strcpy(g_last_rule_violation.reason, + "Public key not whitelisted for this operation"); + return NOSTR_ERROR_AUTH_REQUIRED; } - sqlite3_close(db); return NOSTR_SUCCESS; // Default allow if no restrictive rules matched } diff --git a/src/subscriptions.c b/src/subscriptions.c index d953149..b67e1ed 100644 --- a/src/subscriptions.c +++ b/src/subscriptions.c @@ -1,7 +1,7 @@ #define _GNU_SOURCE #include #include "debug.h" -#include +#include "db_ops.h" #include #include #include @@ -28,9 +28,6 @@ int validate_search_term(const char* search_term, char* error_message, size_t er // Forward declaration for monitoring function void monitoring_on_subscription_change(void); -// Global database variable -extern sqlite3* g_db; - // Configuration functions from config.c extern int get_config_bool(const char* key, int default_value); @@ -991,21 +988,21 @@ int has_subscriptions_for_kind(int event_kind) { // Log subscription creation to database void log_subscription_created(const subscription_t* sub) { - if (!g_db || !sub) return; - + if (!sub) return; + // Convert wsi pointer to string char wsi_str[32]; snprintf(wsi_str, sizeof(wsi_str), "%p", (void*)sub->wsi); - + // Create filter JSON for logging char* filter_json = NULL; if (sub->filters) { cJSON* filters_array = cJSON_CreateArray(); subscription_filter_t* filter = sub->filters; - + while (filter) { cJSON* filter_obj = cJSON_CreateObject(); - + if (filter->kinds) { cJSON_AddItemToObject(filter_obj, "kinds", cJSON_Duplicate(filter->kinds, 1)); } @@ -1034,100 +1031,33 @@ void log_subscription_created(const subscription_t* sub) { } cJSON_Delete(tags_obj); } - + cJSON_AddItemToArray(filters_array, filter_obj); filter = filter->next; } - + filter_json = cJSON_Print(filters_array); cJSON_Delete(filters_array); } - - // Use INSERT OR REPLACE to handle duplicates automatically - const char* sql = - "INSERT OR REPLACE INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type, filter_json) " - "VALUES (?, ?, ?, 'created', ?)"; - - sqlite3_stmt* stmt; - int rc = sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, sub->id, -1, SQLITE_STATIC); - sqlite3_bind_text(stmt, 2, wsi_str, -1, SQLITE_TRANSIENT); - sqlite3_bind_text(stmt, 3, sub->client_ip, -1, SQLITE_STATIC); - sqlite3_bind_text(stmt, 4, filter_json ? filter_json : "[]", -1, SQLITE_TRANSIENT); - - sqlite3_step(stmt); - sqlite3_finalize(stmt); - } - + + db_log_subscription_created(sub->id, wsi_str, sub->client_ip, filter_json ? filter_json : "[]"); + if (filter_json) free(filter_json); } // Log subscription closure to database void log_subscription_closed(const char* sub_id, const char* client_ip, const char* reason) { (void)reason; // Mark as intentionally unused - if (!g_db || !sub_id) return; - - const char* sql = - "INSERT INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type) " - "VALUES (?, '', ?, 'closed')"; - - sqlite3_stmt* stmt; - int rc = sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, sub_id, -1, SQLITE_STATIC); - sqlite3_bind_text(stmt, 2, client_ip ? client_ip : "unknown", -1, SQLITE_STATIC); - - sqlite3_step(stmt); - sqlite3_finalize(stmt); - } - - // Update the corresponding 'created' entry with end time and events sent - const char* update_sql = - "UPDATE subscriptions " - "SET ended_at = strftime('%s', 'now') " - "WHERE subscription_id = ? AND event_type = 'created' AND ended_at IS NULL"; - - rc = sqlite3_prepare_v2(g_db, update_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, sub_id, -1, SQLITE_STATIC); - sqlite3_step(stmt); - sqlite3_finalize(stmt); - } + if (!sub_id) return; + + db_log_subscription_closed(sub_id, client_ip); } // Log subscription disconnection to database void log_subscription_disconnected(const char* client_ip) { - if (!g_db || !client_ip) return; - - // Mark all active subscriptions for this client as disconnected - const char* sql = - "UPDATE subscriptions " - "SET ended_at = strftime('%s', 'now') " - "WHERE client_ip = ? AND event_type = 'created' AND ended_at IS NULL"; - - sqlite3_stmt* stmt; - int rc = sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, client_ip, -1, SQLITE_STATIC); - int changes = sqlite3_changes(g_db); - sqlite3_step(stmt); - sqlite3_finalize(stmt); - - if (changes > 0) { - // Log a disconnection event - const char* insert_sql = - "INSERT INTO subscriptions (subscription_id, wsi_pointer, client_ip, event_type) " - "VALUES ('disconnect', '', ?, 'disconnected')"; - - rc = sqlite3_prepare_v2(g_db, insert_sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_text(stmt, 1, client_ip, -1, SQLITE_STATIC); - sqlite3_step(stmt); - sqlite3_finalize(stmt); - } - } - } + if (!client_ip) return; + + db_log_subscription_disconnected(client_ip); } // Log event broadcast to database (optional, can be resource intensive) @@ -1153,55 +1083,26 @@ void log_subscription_disconnected(const char* client_ip) { // Update events sent counter for a subscription void update_subscription_events_sent(const char* sub_id, int events_sent) { - if (!g_db || !sub_id) return; + if (!sub_id) return; - const char* sql = - "UPDATE subscriptions " - "SET events_sent = ? " - "WHERE subscription_id = ? AND event_type = 'created'"; - - sqlite3_stmt* stmt; - int rc = sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL); - if (rc == SQLITE_OK) { - sqlite3_bind_int(stmt, 1, events_sent); - sqlite3_bind_text(stmt, 2, sub_id, -1, SQLITE_STATIC); - - sqlite3_step(stmt); - sqlite3_finalize(stmt); - } + db_update_subscription_events_sent(sub_id, events_sent); } // Cleanup all subscriptions on startup void cleanup_all_subscriptions_on_startup(void) { - if (!g_db) { + if (!db_is_available()) { DEBUG_ERROR("Database not available for startup cleanup"); return; } - + DEBUG_LOG("Performing startup subscription cleanup"); - - // Mark all active subscriptions as disconnected - const char* sql = - "UPDATE subscriptions " - "SET ended_at = strftime('%s', 'now') " - "WHERE event_type = 'created' AND ended_at IS NULL"; - - sqlite3_stmt* stmt; - int rc = sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL); - if (rc != SQLITE_OK) { - DEBUG_ERROR("Failed to prepare startup cleanup query"); - return; - } - - rc = sqlite3_step(stmt); - int changes = sqlite3_changes(g_db); - sqlite3_finalize(stmt); - - if (rc != SQLITE_DONE) { + + int changes = db_cleanup_orphaned_subscriptions(); + if (changes < 0) { DEBUG_ERROR("Failed to execute startup cleanup"); return; } - + if (changes > 0) { DEBUG_LOG("Startup cleanup: marked %d orphaned subscriptions as disconnected", changes); } else { diff --git a/src/thread_pool.c b/src/thread_pool.c new file mode 100644 index 0000000..b9bc966 --- /dev/null +++ b/src/thread_pool.c @@ -0,0 +1,276 @@ +#define _GNU_SOURCE + +#include "thread_pool.h" +#include "debug.h" + +#include +#include +#include + +typedef struct thread_pool_job_node { + uint64_t job_id; + thread_pool_job_t job; + struct thread_pool_job_node* next; +} thread_pool_job_node_t; + +typedef struct { + thread_pool_job_node_t* head; + thread_pool_job_node_t* tail; + int size; + int max_size; + pthread_mutex_t mutex; + pthread_cond_t cond; +} thread_pool_queue_t; + +typedef struct { + int running; + uint64_t next_job_id; + + int reader_count; + pthread_t* readers; + pthread_t writer; + + thread_pool_queue_t read_q; + thread_pool_queue_t write_q; + + thread_pool_wake_loop_cb wake_loop_cb; + void* wake_loop_ctx; + + pthread_mutex_t state_mutex; +} thread_pool_state_t; + +static thread_pool_state_t g_pool = {0}; + +static void queue_init(thread_pool_queue_t* q, int max_size) { + memset(q, 0, sizeof(*q)); + q->max_size = (max_size > 0) ? max_size : 4096; + pthread_mutex_init(&q->mutex, NULL); + pthread_cond_init(&q->cond, NULL); +} + +static void queue_destroy(thread_pool_queue_t* q) { + pthread_mutex_lock(&q->mutex); + thread_pool_job_node_t* cur = q->head; + while (cur) { + thread_pool_job_node_t* next = cur->next; + if (cur->job.payload_free && cur->job.payload) { + cur->job.payload_free(cur->job.payload); + } + free(cur); + cur = next; + } + q->head = q->tail = NULL; + q->size = 0; + pthread_mutex_unlock(&q->mutex); + + pthread_mutex_destroy(&q->mutex); + pthread_cond_destroy(&q->cond); +} + +static int queue_push(thread_pool_queue_t* q, thread_pool_job_node_t* node) { + pthread_mutex_lock(&q->mutex); + if (q->size >= q->max_size) { + pthread_mutex_unlock(&q->mutex); + return 0; + } + + node->next = NULL; + if (!q->tail) { + q->head = q->tail = node; + } else { + q->tail->next = node; + q->tail = node; + } + q->size++; + pthread_cond_signal(&q->cond); + pthread_mutex_unlock(&q->mutex); + return 1; +} + +static thread_pool_job_node_t* queue_pop(thread_pool_queue_t* q) { + pthread_mutex_lock(&q->mutex); + while (g_pool.running && q->size == 0) { + pthread_cond_wait(&q->cond, &q->mutex); + } + + if (!g_pool.running && q->size == 0) { + pthread_mutex_unlock(&q->mutex); + return NULL; + } + + thread_pool_job_node_t* node = q->head; + q->head = node->next; + if (!q->head) q->tail = NULL; + q->size--; + pthread_mutex_unlock(&q->mutex); + return node; +} + +static void wake_event_loop(void) { + if (g_pool.wake_loop_cb) { + g_pool.wake_loop_cb(g_pool.wake_loop_ctx); + } +} + +static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_status_t status, const char* message) { + thread_pool_result_t result; + memset(&result, 0, sizeof(result)); + result.job_id = node->job_id; + result.type = node->job.type; + result.status = status; + result.session = node->job.session; + result.message = message; + + if (node->job.result_cb) { + node->job.result_cb(&result, node->job.result_cb_ctx); + } + wake_event_loop(); + + if (node->job.payload_free && node->job.payload) { + node->job.payload_free(node->job.payload); + } + free(node); +} + +static void* reader_worker_main(void* arg) { + (void)arg; + while (g_pool.running) { + thread_pool_job_node_t* node = queue_pop(&g_pool.read_q); + if (!node) break; + + // Scaffold only: execution wiring is intentionally deferred. + complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, + "Read worker scaffold active; execution not wired yet"); + } + return NULL; +} + +static void* writer_worker_main(void* arg) { + (void)arg; + while (g_pool.running) { + thread_pool_job_node_t* node = queue_pop(&g_pool.write_q); + if (!node) break; + + // Scaffold only: execution wiring is intentionally deferred. + complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED, + "Write worker scaffold active; execution not wired yet"); + } + return NULL; +} + +int thread_pool_init(const thread_pool_config_t* config) { + if (!config) return -1; + + pthread_mutex_lock(&g_pool.state_mutex); + if (g_pool.running) { + pthread_mutex_unlock(&g_pool.state_mutex); + return 0; + } + + g_pool.reader_count = (config->reader_threads > 0) ? config->reader_threads : 4; + g_pool.readers = calloc((size_t)g_pool.reader_count, sizeof(pthread_t)); + if (!g_pool.readers) { + pthread_mutex_unlock(&g_pool.state_mutex); + return -1; + } + + queue_init(&g_pool.read_q, config->max_queue_depth); + queue_init(&g_pool.write_q, config->max_queue_depth); + + g_pool.next_job_id = 1; + g_pool.wake_loop_cb = config->wake_loop_cb; + g_pool.wake_loop_ctx = config->wake_loop_ctx; + g_pool.running = 1; + + for (int i = 0; i < g_pool.reader_count; i++) { + if (pthread_create(&g_pool.readers[i], NULL, reader_worker_main, NULL) != 0) { + g_pool.running = 0; + pthread_cond_broadcast(&g_pool.read_q.cond); + pthread_cond_broadcast(&g_pool.write_q.cond); + pthread_mutex_unlock(&g_pool.state_mutex); + return -1; + } + } + + if (pthread_create(&g_pool.writer, NULL, writer_worker_main, NULL) != 0) { + g_pool.running = 0; + pthread_cond_broadcast(&g_pool.read_q.cond); + pthread_cond_broadcast(&g_pool.write_q.cond); + pthread_mutex_unlock(&g_pool.state_mutex); + return -1; + } + + pthread_mutex_unlock(&g_pool.state_mutex); + DEBUG_LOG("Thread pool initialized: %d readers + 1 writer", g_pool.reader_count); + return 0; +} + +void thread_pool_shutdown(void) { + pthread_mutex_lock(&g_pool.state_mutex); + if (!g_pool.running) { + pthread_mutex_unlock(&g_pool.state_mutex); + return; + } + + g_pool.running = 0; + pthread_cond_broadcast(&g_pool.read_q.cond); + pthread_cond_broadcast(&g_pool.write_q.cond); + pthread_mutex_unlock(&g_pool.state_mutex); + + for (int i = 0; i < g_pool.reader_count; i++) { + pthread_join(g_pool.readers[i], NULL); + } + pthread_join(g_pool.writer, NULL); + + free(g_pool.readers); + g_pool.readers = NULL; + g_pool.reader_count = 0; + + queue_destroy(&g_pool.read_q); + queue_destroy(&g_pool.write_q); + + DEBUG_LOG("Thread pool shutdown complete"); +} + +int thread_pool_is_running(void) { + return g_pool.running; +} + +static thread_pool_status_t submit_to_queue(thread_pool_queue_t* q, const thread_pool_job_t* job, uint64_t* out_job_id) { + if (!g_pool.running) return THREAD_POOL_STATUS_SHUTTING_DOWN; + if (!job) return THREAD_POOL_STATUS_INTERNAL_ERROR; + + thread_pool_job_node_t* node = calloc(1, sizeof(*node)); + if (!node) return THREAD_POOL_STATUS_INTERNAL_ERROR; + + node->job = *job; + + pthread_mutex_lock(&g_pool.state_mutex); + node->job_id = g_pool.next_job_id++; + pthread_mutex_unlock(&g_pool.state_mutex); + + if (out_job_id) *out_job_id = node->job_id; + + if (!queue_push(q, node)) { + if (job->payload_free && job->payload) { + job->payload_free(job->payload); + } + free(node); + return THREAD_POOL_STATUS_QUEUE_FULL; + } + + return THREAD_POOL_STATUS_OK; +} + +thread_pool_status_t thread_pool_submit_read(const thread_pool_job_t* job, uint64_t* out_job_id) { + return submit_to_queue(&g_pool.read_q, job, out_job_id); +} + +thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint64_t* out_job_id) { + return submit_to_queue(&g_pool.write_q, job, out_job_id); +} + +__attribute__((constructor)) +static void thread_pool_state_init_once(void) { + pthread_mutex_init(&g_pool.state_mutex, NULL); +} diff --git a/src/thread_pool.h b/src/thread_pool.h new file mode 100644 index 0000000..6b652d1 --- /dev/null +++ b/src/thread_pool.h @@ -0,0 +1,70 @@ +#ifndef THREAD_POOL_H +#define THREAD_POOL_H + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef enum { + THREAD_POOL_JOB_REQ_QUERY = 0, + THREAD_POOL_JOB_COUNT_QUERY, + THREAD_POOL_JOB_STORE_EVENT, + THREAD_POOL_JOB_DELETE_EVENT, + THREAD_POOL_JOB_CUSTOM +} thread_pool_job_type_t; + +typedef enum { + THREAD_POOL_STATUS_OK = 0, + THREAD_POOL_STATUS_NOT_IMPLEMENTED, + THREAD_POOL_STATUS_QUEUE_FULL, + THREAD_POOL_STATUS_SHUTTING_DOWN, + THREAD_POOL_STATUS_INTERNAL_ERROR +} thread_pool_status_t; + +typedef struct { + uint64_t job_id; + thread_pool_job_type_t type; + thread_pool_status_t status; + void* session; + void* result_data; + size_t result_size; + const char* message; +} thread_pool_result_t; + +typedef void (*thread_pool_result_cb)(const thread_pool_result_t* result, void* user_ctx); +typedef void (*thread_pool_payload_free_cb)(void* payload); +typedef void (*thread_pool_wake_loop_cb)(void* wake_ctx); + +typedef struct { + thread_pool_job_type_t type; + void* session; + void* payload; + size_t payload_size; + thread_pool_payload_free_cb payload_free; + thread_pool_result_cb result_cb; + void* result_cb_ctx; +} thread_pool_job_t; + +typedef struct { + int reader_threads; + int max_queue_depth; + const char* db_path; + thread_pool_wake_loop_cb wake_loop_cb; + void* wake_loop_ctx; +} thread_pool_config_t; + +int thread_pool_init(const thread_pool_config_t* config); +void thread_pool_shutdown(void); +int thread_pool_is_running(void); + +thread_pool_status_t thread_pool_submit_read(const thread_pool_job_t* job, uint64_t* out_job_id); +thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint64_t* out_job_id); + +#ifdef __cplusplus +} +#endif + +#endif // THREAD_POOL_H diff --git a/src/websockets.c b/src/websockets.c index 31e59e1..8562674 100644 --- a/src/websockets.c +++ b/src/websockets.c @@ -31,6 +31,7 @@ #include "api.h" // API for embedded files #include "dm_admin.h" // DM admin functions including NIP-17 #include "ip_ban.h" // IP auth failure ban system +#include "thread_pool.h" // Thread pool scaffold // Forward declarations for logging functions @@ -2454,8 +2455,7 @@ static void check_connection_age(int max_connection_seconds) { // Periodic IP ban maintenance: cleanup expired entries, log stats, save to DB ip_ban_cleanup(); - extern sqlite3* g_db; - ip_ban_log_stats(g_db); + ip_ban_log_stats(); } // WebSocket protocol definition @@ -2676,8 +2676,7 @@ int start_websocket_relay(int port_override, int strict_port) { } else { // Even when connection age limit is disabled, run IP ban maintenance ip_ban_cleanup(); - extern sqlite3* g_db; - ip_ban_log_stats(g_db); + ip_ban_log_stats(); } } } @@ -2863,6 +2862,10 @@ int process_dm_stats_command(cJSON* dm_event, char* error_message, size_t error_ int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { // pss is now used for query tracking, so remove unused warning suppression + if (thread_pool_is_running()) { + DEBUG_TRACE("Thread pool scaffold active: handle_count_message() still using synchronous DB path"); + } + if (!cJSON_IsArray(filters)) { DEBUG_ERROR("COUNT filters is not an array"); return 0;