laantungir/c-relay-pg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.0.4 - Complete Phase 1 DB abstraction: opaque db_stmt API, ip_ban migration, and SQLite leakage cleanup

Browse Source
This commit is contained in:
Laan Tungir
2026-04-01 09:10:42 -04:00
parent bc9ac290ab
commit cbd260c1ae
9 changed files with 355 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
postgres_relays.md
+166
View File
@@ -6,6 +6,8 @@ This document summarizes common approaches and best practices for building Nostr
* **Nostrss**: High-performance Rust-based relay.
* **Nostrgres**: Focused on PostgreSQL integration with complex event filtering.
* **Relay-rs (Postgres branch)**: General purpose high-throughput relay.
* **Nostream** ([github.com/Cameri/nostream](https://github.com/Cameri/nostream)): Production-ready TypeScript relay backed by PostgreSQL and Redis.
* **Nostr-Relay-NestJS** ([github.com/CodyTseng/nostr-relay-nestjs](https://github.com/CodyTseng/nostr-relay-nestjs)): High-performance relay built with NestJS and PostgreSQL, utilizing Kysely for type-safe database interactions.
## Detailed Analysis
@@ -29,6 +31,125 @@ This document summarizes common approaches and best practices for building Nostr
3. **Connection Pooling**: PostgreSQL requires aggressive connection management. Use a high-performance pooler like `PgBouncer` to handle the large number of concurrent, short-lived connections common in WebSocket-based relay traffic.
4. **Asynchronous Writes**: Decouple the WebSocket ingestion thread from the database writer thread to ensure that slow database writes do not impact the relay's responsiveness.
## Nostream ([github.com/Cameri/nostream](https://github.com/Cameri/nostream))
A production-ready Nostr relay written in TypeScript (Node.js), backed by PostgreSQL 14+ and Redis. Uses Knex.js for versioned migrations.
### PostgreSQL Schema
The schema is fully normalized — no JSONB blob for the whole event. Tags are stored separately and populated by a trigger.
**`events` table** — one row per event, tags kept as a JSONB column for the trigger to process:
```sql
CREATE TABLE events (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
event_id BYTEA UNIQUE NOT NULL, -- 32-byte hash
event_pubkey BYTEA NOT NULL,
event_kind INTEGER UNSIGNED NOT NULL,
event_created_at INTEGER UNSIGNED NOT NULL,
event_content TEXT NOT NULL,
event_tags JSONB, -- raw tags array, source of truth for trigger
event_signature BYTEA NOT NULL,
remote_address TEXT,
expires_at INTEGER,
deleted_at TIMESTAMP,
event_deduplication JSONB, -- used for replaceable event conflict key
first_seen TIMESTAMP DEFAULT now()
);
-- Indexes
CREATE INDEX ON events (event_pubkey);
CREATE INDEX ON events (event_kind);
CREATE INDEX ON events (event_created_at);
CREATE INDEX ON events (expires_at);
CREATE INDEX event_tags_idx ON events USING GIN (event_tags); -- GIN on raw JSONB
-- Unique index enforcing NIP-16 replaceable event semantics
CREATE UNIQUE INDEX replaceable_events_idx ON events (event_pubkey, event_kind)
WHERE event_kind = 0 OR event_kind = 3
OR (event_kind >= 10000 AND event_kind < 20000);
```
**`event_tags` table** — denormalized single-letter tags, populated by a PostgreSQL trigger on `events`:
```sql
CREATE TABLE event_tags (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
event_id BYTEA NOT NULL,
tag_name TEXT NOT NULL, -- single-letter only (e.g. 'e', 'p', 't')
tag_value TEXT NOT NULL
);
CREATE INDEX ON event_tags (event_id);
CREATE INDEX ON event_tags (tag_name, tag_value);
-- Trigger: on INSERT/UPDATE/DELETE to events, repopulate event_tags
CREATE OR REPLACE FUNCTION process_event_tags() RETURNS TRIGGER AS $$
DECLARE
tag_element jsonb;
BEGIN
DELETE FROM event_tags WHERE event_id = OLD.event_id;
IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
FOR tag_element IN SELECT jsonb_array_elements(NEW.event_tags) LOOP
IF length(trim('"' FROM (tag_element->0)::text)) = 1
AND (tag_element->1)::text IS NOT NULL THEN
INSERT INTO event_tags (event_id, tag_name, tag_value)
VALUES (NEW.event_id,
trim('"' FROM (tag_element->0)::text),
trim('"' FROM (tag_element->1)::text));
END IF;
END LOOP;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER insert_event_tags
AFTER INSERT OR UPDATE OR DELETE ON events
FOR EACH ROW EXECUTE FUNCTION process_event_tags();
```
Tag queries (`#e`, `#p`, etc.) join `event_tags` rather than traversing the JSONB array.
**`users` and `invoices` tables** — for paid relay admission (Lightning payments):
```sql
CREATE TABLE users (
pubkey BYTEA PRIMARY KEY,
is_admitted BOOLEAN DEFAULT FALSE,
balance BIGINT DEFAULT 0, -- in msats
tos_accepted_at DATETIME
);
CREATE TABLE invoices (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
pubkey BYTEA NOT NULL,
bolt11 TEXT NOT NULL,
amount_requested BIGINT UNSIGNED NOT NULL,
amount_paid BIGINT UNSIGNED,
unit ENUM('msats','sats','btc'),
status ENUM('pending','completed','expired'),
description TEXT,
confirmed_at DATETIME,
expires_at DATETIME
);
```
### How Redis Is Used
Redis is **not** used for subscription fan-out. Subscription delivery is handled entirely in-process: each `WebSocketAdapter` holds a `Map<subscriptionId, filters[]>`, and a broadcast event walks all connected clients and filters locally.
Redis has two actual roles:
1. **Sliding-window rate limiting** — The `SlidingWindowRateLimiter` uses Redis sorted sets to track request timestamps per key (pubkey, IP, etc.) within a rolling time window:
- `ZREMRANGEBYSCORE key 0 (now - period)` — evict old entries
- `ZADD key timestamp member` — record this hit
- `ZRANGE key 0 -1` — count hits in window
- `EXPIRE key period` — auto-cleanup
- Keys follow the pattern `<pubkey>:events:<period>` or `<ip>:message:<period>`
2. **General-purpose cache** — A `RedisAdapter` wraps the client with `get`/`set`/`hasKey` helpers, used for caching admission checks (e.g. `<pubkey>:is-admitted`). The event handler has a `TODO` comment noting that the `userRepository.findByPubkey()` call should be replaced with a cache lookup — meaning this is partially implemented.
---
## Schema Definitions
### Relay-rs (Postgres)
@@ -89,3 +210,48 @@ CREATE TABLE "invoice" (
CONSTRAINT invoice_pubkey_fkey FOREIGN KEY (pubkey) REFERENCES account (pubkey) ON DELETE CASCADE
);
```
### Nostr-Relay-NestJS (PostgreSQL via Kysely)
```sql
-- Events table
CREATE TABLE events (
id CHAR(64) PRIMARY KEY,
pubkey CHAR(64) NOT NULL,
created_at BIGINT NOT NULL,
kind INTEGER NOT NULL,
tags JSONB NOT NULL DEFAULT '[]',
generic_tags TEXT[] NOT NULL DEFAULT '{}',
content TEXT NOT NULL DEFAULT '',
sig CHAR(128) NOT NULL,
expired_at BIGINT,
d_tag_value TEXT,
delegator CHAR(64),
create_date TIMESTAMP NOT NULL DEFAULT now(),
update_date TIMESTAMP NOT NULL DEFAULT now(),
delete_date TIMESTAMP
);
-- Indices
CREATE EXTENSION IF NOT EXISTS btree_gin;
CREATE INDEX generic_tags_kind_idx ON events USING gin (generic_tags, kind);
CREATE INDEX pubkey_kind_created_at_idx ON events (pubkey, kind, created_at);
CREATE INDEX delegator_kind_created_at_idx ON events (delegator, kind, created_at);
CREATE INDEX created_at_kind_idx ON events (created_at, kind);
-- Generic tags table (for optimized tag queries)
CREATE TABLE generic_tags (
id SERIAL PRIMARY KEY,
tag TEXT NOT NULL,
author CHAR(64) NOT NULL,
kind INTEGER NOT NULL,
event_id CHAR(64) NOT NULL REFERENCES events(id),
created_at BIGINT NOT NULL
);
-- Indices for tag filtering
CREATE UNIQUE INDEX g_tag_event_id_idx ON generic_tags (tag, event_id);
CREATE INDEX g_tag_kind_created_at_idx ON generic_tags (tag, kind, created_at);
CREATE INDEX g_tag_created_at_idx ON generic_tags (tag, created_at);
```
Nostr-Relay-NestJS features a hybrid storage model where tags are stored both in a JSONB field (for general lookup) and a normalized `generic_tags` table (for optimized tag query performance). The schema uses Kysely for type-safe interaction.
relay.pid
+1 -1
View File
@@ -1 +1 @@
484178
504823
src/config.c
+55 -55
View File
@@ -2067,7 +2067,7 @@ int wot_sync_from_admin_kind3(void) {
DEBUG_INFO("WoT sync: Starting sync from admin's kind 3 event (level %d)", wot_level);
sqlite3_stmt* stmt = NULL;
db_stmt_t* stmt = NULL;
int rc;
int whitelisted_count = 0;
@@ -2079,7 +2079,7 @@ int wot_sync_from_admin_kind3(void) {
"AND e.created_at = (SELECT MAX(created_at) FROM events WHERE kind = 3 AND pubkey = ?)";
rc = db_prepare(query_sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to prepare p-tag query: %s", db_last_error());
free((char*)admin_pubkey);
free((char*)relay_pubkey);
@@ -2103,7 +2103,7 @@ int wot_sync_from_admin_kind3(void) {
return -1;
}
while ((rc = db_step_stmt(stmt)) == SQLITE_ROW) {
while ((rc = db_step_stmt(stmt)) == DB_ROW) {
const char* p_tag = (const char*)db_column_text_value(stmt, 0);
if (p_tag) {
// Expand array if needed
@@ -2139,7 +2139,7 @@ int wot_sync_from_admin_kind3(void) {
// Step 2: Begin transaction
rc = db_exec_sql("BEGIN TRANSACTION");
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to begin transaction: %s", db_last_error());
for (int i = 0; i < p_tag_count; i++) {
free(p_tags[i]);
@@ -2170,7 +2170,7 @@ int wot_sync_from_admin_kind3(void) {
"VALUES ('wot_whitelist', 'pubkey', ?, 1)";
rc = db_prepare(insert_sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to prepare insert statement: %s", db_last_error());
db_exec_sql("ROLLBACK");
for (int i = 0; i < p_tag_count; i++) {
@@ -2186,7 +2186,7 @@ int wot_sync_from_admin_kind3(void) {
rc = db_step_stmt(stmt);
db_finalize_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("WoT sync: Failed to insert admin pubkey whitelist: %s", db_last_error());
db_exec_sql("ROLLBACK");
for (int i = 0; i < p_tag_count; i++) {
@@ -2202,7 +2202,7 @@ int wot_sync_from_admin_kind3(void) {
// Step 5: Insert relay pubkey as wot_whitelist (needed for DM responses)
rc = db_prepare(insert_sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to prepare insert statement for relay: %s", db_last_error());
db_exec_sql("ROLLBACK");
for (int i = 0; i < p_tag_count; i++) {
@@ -2218,7 +2218,7 @@ int wot_sync_from_admin_kind3(void) {
rc = db_step_stmt(stmt);
db_finalize_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("WoT sync: Failed to insert relay pubkey whitelist: %s", db_last_error());
db_exec_sql("ROLLBACK");
for (int i = 0; i < p_tag_count; i++) {
@@ -2235,7 +2235,7 @@ int wot_sync_from_admin_kind3(void) {
// Step 6: Insert each p tag from kind 3 event
for (int i = 0; i < p_tag_count; i++) {
rc = db_prepare(insert_sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to prepare insert for p_tag %d: %s", i, db_last_error());
db_exec_sql("ROLLBACK");
// Free remaining tags
@@ -2252,7 +2252,7 @@ int wot_sync_from_admin_kind3(void) {
rc = db_step_stmt(stmt);
db_finalize_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("WoT sync: Failed to insert p_tag %d: %s", i, db_last_error());
db_exec_sql("ROLLBACK");
// Free remaining tags
@@ -2269,7 +2269,7 @@ int wot_sync_from_admin_kind3(void) {
// Step 7: Commit transaction
rc = db_exec_sql("COMMIT");
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("WoT sync: Failed to commit transaction: %s", db_last_error());
db_exec_sql("ROLLBACK");
for (int i = 0; i < p_tag_count; i++) {
@@ -2973,9 +2973,9 @@ int handle_auth_query_unified(cJSON* event, const char* query_type, char* error_
}
// Execute query
sqlite3_stmt* stmt;
db_stmt_t* stmt;
int rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
snprintf(error_message, error_size, "failed to prepare auth query");
return -1;
}
@@ -2995,7 +2995,7 @@ int handle_auth_query_unified(cJSON* event, const char* query_type, char* error_
int rule_count = 0;
// printf("=== Auth Query Results (%s) ===\n", query_type);
while (db_step_stmt(stmt) == SQLITE_ROW) {
while (db_step_stmt(stmt) == DB_ROW) {
const char* rule_type = (const char*)db_column_text_value(stmt, 0);
const char* pattern_type = (const char*)db_column_text_value(stmt, 1);
const char* pattern_value_result = (const char*)db_column_text_value(stmt, 2);
@@ -3091,9 +3091,9 @@ int handle_config_query_unified(cJSON* event, const char* query_type, char* erro
}
// Execute query
sqlite3_stmt* stmt;
db_stmt_t* stmt;
int rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
snprintf(error_message, error_size, "failed to prepare config query");
return -1;
}
@@ -3112,7 +3112,7 @@ int handle_config_query_unified(cJSON* event, const char* query_type, char* erro
int config_count = 0;
while (db_step_stmt(stmt) == SQLITE_ROW) {
while (db_step_stmt(stmt) == DB_ROW) {
const char* key = (const char*)db_column_text_value(stmt, 0);
const char* value = (const char*)db_column_text_value(stmt, 1);
const char* data_type = (const char*)db_column_text_value(stmt, 2);
@@ -3680,9 +3680,9 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
cJSON_AddNumberToObject(response, "database_size_bytes", db_size);
// Query total events count
sqlite3_stmt* stmt;
if (db_prepare("SELECT COUNT(*) FROM events", &stmt) == SQLITE_OK) {
if (db_step_stmt(stmt) == SQLITE_ROW) {
db_stmt_t* stmt;
if (db_prepare("SELECT COUNT(*) FROM events", &stmt) == DB_OK) {
if (db_step_stmt(stmt) == DB_ROW) {
cJSON_AddNumberToObject(response, "total_events", db_column_int64_value(stmt, 0));
}
db_finalize_stmt(stmt);
@@ -3690,8 +3690,8 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
// Query event kinds distribution
cJSON* event_kinds = cJSON_CreateArray();
if (db_prepare("SELECT kind, count, percentage FROM event_kinds_view ORDER BY count DESC", &stmt) == SQLITE_OK) {
while (db_step_stmt(stmt) == SQLITE_ROW) {
if (db_prepare("SELECT kind, count, percentage FROM event_kinds_view ORDER BY count DESC", &stmt) == DB_OK) {
while (db_step_stmt(stmt) == DB_ROW) {
cJSON* kind_obj = cJSON_CreateObject();
cJSON_AddNumberToObject(kind_obj, "kind", db_column_int_value(stmt, 0));
cJSON_AddNumberToObject(kind_obj, "count", db_column_int64_value(stmt, 1));
@@ -3704,10 +3704,10 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
// Query time-based statistics
cJSON* time_stats = cJSON_CreateObject();
if (db_prepare("SELECT period, total_events FROM time_stats_view", &stmt) == SQLITE_OK) {
while (db_step_stmt(stmt) == SQLITE_ROW) {
if (db_prepare("SELECT period, total_events FROM time_stats_view", &stmt) == DB_OK) {
while (db_step_stmt(stmt) == DB_ROW) {
const char* period = (const char*)db_column_text_value(stmt, 0);
sqlite3_int64 count = db_column_int64_value(stmt, 1);
long long count = db_column_int64_value(stmt, 1);
if (strcmp(period, "total") == 0) {
cJSON_AddNumberToObject(time_stats, "total", count);
@@ -3725,8 +3725,8 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
// Query top pubkeys
cJSON* top_pubkeys = cJSON_CreateArray();
if (db_prepare("SELECT pubkey, event_count, percentage FROM top_pubkeys_view ORDER BY event_count DESC LIMIT 10", &stmt) == SQLITE_OK) {
while (db_step_stmt(stmt) == SQLITE_ROW) {
if (db_prepare("SELECT pubkey, event_count, percentage FROM top_pubkeys_view ORDER BY event_count DESC LIMIT 10", &stmt) == DB_OK) {
while (db_step_stmt(stmt) == DB_ROW) {
cJSON* pubkey_obj = cJSON_CreateObject();
const char* pubkey = (const char*)db_column_text_value(stmt, 0);
cJSON_AddStringToObject(pubkey_obj, "pubkey", pubkey ? pubkey : "");
@@ -3739,9 +3739,9 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
cJSON_AddItemToObject(response, "top_pubkeys", top_pubkeys);
// Get database creation timestamp (oldest event)
if (db_prepare("SELECT MIN(created_at) FROM events", &stmt) == SQLITE_OK) {
if (db_step_stmt(stmt) == SQLITE_ROW) {
sqlite3_int64 oldest_timestamp = db_column_int64_value(stmt, 0);
if (db_prepare("SELECT MIN(created_at) FROM events", &stmt) == DB_OK) {
if (db_step_stmt(stmt) == DB_ROW) {
long long oldest_timestamp = db_column_int64_value(stmt, 0);
if (oldest_timestamp > 0) {
cJSON_AddNumberToObject(response, "database_created_at", (double)oldest_timestamp);
}
@@ -3750,9 +3750,9 @@ int handle_stats_query_unified(cJSON* event, char* error_message, size_t error_s
}
// Get latest event timestamp
if (db_prepare("SELECT MAX(created_at) FROM events", &stmt) == SQLITE_OK) {
if (db_step_stmt(stmt) == SQLITE_ROW) {
sqlite3_int64 latest_timestamp = db_column_int64_value(stmt, 0);
if (db_prepare("SELECT MAX(created_at) FROM events", &stmt) == DB_OK) {
if (db_step_stmt(stmt) == DB_ROW) {
long long latest_timestamp = db_column_int64_value(stmt, 0);
if (latest_timestamp > 0) {
cJSON_AddNumberToObject(response, "latest_event_at", (double)latest_timestamp);
}
@@ -3884,7 +3884,7 @@ int handle_config_update_unified(cJSON* event, char* error_message, size_t error
// Begin transaction for atomic config updates
int rc = db_exec_sql("BEGIN IMMEDIATE TRANSACTION");
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
cJSON_Delete(config_objects_array);
snprintf(error_message, error_size, "failed to begin config update transaction");
return -1;
@@ -3960,10 +3960,10 @@ int handle_config_update_unified(cJSON* event, char* error_message, size_t error
// Check if the config key exists in the table
const char* check_sql = "SELECT COUNT(*) FROM config WHERE key = ?";
sqlite3_stmt* check_stmt;
db_stmt_t* check_stmt;
int check_rc = db_prepare(check_sql, &check_stmt);
if (check_rc != SQLITE_OK) {
if (check_rc != DB_OK) {
DEBUG_ERROR("Failed to prepare config existence check");
validation_errors++;
continue;
@@ -3972,7 +3972,7 @@ int handle_config_update_unified(cJSON* event, char* error_message, size_t error
db_bind_text_param(check_stmt, 1, key);
int config_exists = 0;
if (db_step_stmt(check_stmt) == SQLITE_ROW) {
if (db_step_stmt(check_stmt) == DB_ROW) {
config_exists = db_column_int_value(check_stmt, 0) > 0;
}
db_finalize_stmt(check_stmt);
@@ -3996,12 +3996,12 @@ int handle_config_update_unified(cJSON* event, char* error_message, size_t error
// Check if this config requires restart
const char* requires_restart_sql = "SELECT requires_restart FROM config WHERE key = ?";
sqlite3_stmt* restart_stmt;
db_stmt_t* restart_stmt;
int requires_restart = 0;
if (db_prepare(requires_restart_sql, &restart_stmt) == SQLITE_OK) {
if (db_prepare(requires_restart_sql, &restart_stmt) == DB_OK) {
db_bind_text_param(restart_stmt, 1, key);
if (db_step_stmt(restart_stmt) == SQLITE_ROW) {
if (db_step_stmt(restart_stmt) == DB_ROW) {
requires_restart = db_column_int_value(restart_stmt, 0);
}
db_finalize_stmt(restart_stmt);
@@ -4295,10 +4295,10 @@ int populate_all_config_values_atomic(const char* admin_pubkey, const char* rela
}
// Prepare INSERT OR REPLACE statement with all required fields
sqlite3_stmt* stmt = NULL;
db_stmt_t* stmt = NULL;
const char* sql = "INSERT OR REPLACE INTO config (key, value, data_type, description, category, requires_restart) VALUES (?, ?, ?, ?, ?, ?)";
rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("Failed to prepare statement: %s", db_last_error());
db_exec_sql("ROLLBACK;");
return -1;
@@ -4378,7 +4378,7 @@ int populate_all_config_values_atomic(const char* admin_pubkey, const char* rela
db_bind_int_param(stmt, 6, requires_restart);
rc = db_step_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("Failed to insert config key '%s': %s",
key, db_last_error());
db_finalize_stmt(stmt);
@@ -4396,7 +4396,7 @@ int populate_all_config_values_atomic(const char* admin_pubkey, const char* rela
db_bind_text_param(stmt, 5, "authentication");
db_bind_int_param(stmt, 6, 0); // does not require restart
rc = db_step_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("Failed to insert admin_pubkey: %s", db_last_error());
db_finalize_stmt(stmt);
db_exec_sql("ROLLBACK;");
@@ -4412,7 +4412,7 @@ int populate_all_config_values_atomic(const char* admin_pubkey, const char* rela
db_bind_text_param(stmt, 5, "relay");
db_bind_int_param(stmt, 6, 0); // does not require restart
rc = db_step_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("Failed to insert relay_pubkey: %s", db_last_error());
db_finalize_stmt(stmt);
db_exec_sql("ROLLBACK;");
@@ -4429,7 +4429,7 @@ int populate_all_config_values_atomic(const char* admin_pubkey, const char* rela
db_bind_text_param(stmt, 5, "monitoring");
db_bind_int_param(stmt, 6, 0); // does not require restart
rc = db_step_stmt(stmt);
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
DEBUG_ERROR("Failed to insert kind_24567_reporting_throttle_sec: %s", db_last_error());
db_finalize_stmt(stmt);
db_exec_sql("ROLLBACK;");
@@ -4483,15 +4483,15 @@ int is_config_table_ready(void) {
if (!db_is_available()) return 0;
const char* sql = "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='config'";
sqlite3_stmt* stmt;
db_stmt_t* stmt;
int rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
return 0;
}
int table_exists = 0;
if (db_step_stmt(stmt) == SQLITE_ROW) {
if (db_step_stmt(stmt) == DB_ROW) {
table_exists = db_column_int_value(stmt, 0) > 0;
}
db_finalize_stmt(stmt);
@@ -4503,12 +4503,12 @@ int is_config_table_ready(void) {
// Check if table has configuration data
const char* count_sql = "SELECT COUNT(*) FROM config";
rc = db_prepare(count_sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
return 0;
}
int config_count = 0;
if (db_step_stmt(stmt) == SQLITE_ROW) {
if (db_step_stmt(stmt) == DB_ROW) {
config_count = db_column_int_value(stmt, 0);
}
db_finalize_stmt(stmt);
@@ -4731,7 +4731,7 @@ int process_startup_config_event(const cJSON* event) {
// Begin transaction for atomic config updates
int rc = db_exec_sql("BEGIN IMMEDIATE TRANSACTION");
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("Failed to begin startup config transaction");
return -1;
}
@@ -4867,10 +4867,10 @@ cJSON* generate_config_event_from_table(void) {
// Query all configuration values from the config table
const char* sql = "SELECT key, value FROM config ORDER BY key";
sqlite3_stmt* stmt;
db_stmt_t* stmt;
int rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("Failed to prepare config query for event generation");
cJSON_Delete(tags);
cJSON_Delete(event);
@@ -4880,7 +4880,7 @@ cJSON* generate_config_event_from_table(void) {
int config_items_added = 0;
// Add each config item as a tag
while (db_step_stmt(stmt) == SQLITE_ROW) {
while (db_step_stmt(stmt) == DB_ROW) {
const char* key = (const char*)db_column_text_value(stmt, 0);
const char* value = (const char*)db_column_text_value(stmt, 1);
src/config.h
-1
View File
@@ -1,7 +1,6 @@
#ifndef CONFIG_H
#define CONFIG_H
#include <sqlite3.h>
#include <cjson/cJSON.h>
#include <time.h>
#include <pthread.h>
src/db_ops.c
+52 -36
View File
@@ -4,6 +4,7 @@
#include "debug.h"
#include "config.h"
#include <sqlite3.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -12,14 +13,14 @@
extern sqlite3* g_db;
extern char g_database_path[512];
typedef struct db_stmt {
sqlite3_stmt* stmt;
} db_stmt_t;
int db_is_available(void) {
return g_db != NULL;
}
sqlite3* db_get_handle(void) {
return g_db;
}
const char* db_last_error(void) {
if (!g_db) return "database not available";
return sqlite3_errmsg(g_db);
@@ -29,58 +30,73 @@ const char* db_get_database_path(void) {
return g_database_path;
}
int db_prepare(const char* sql, sqlite3_stmt** out_stmt) {
if (!g_db || !sql || !out_stmt) return SQLITE_MISUSE;
return sqlite3_prepare_v2(g_db, sql, -1, out_stmt, NULL);
int db_prepare(const char* sql, db_stmt_t** out_stmt) {
if (!g_db || !sql || !out_stmt) return DB_MISUSE;
sqlite3_stmt* raw_stmt = NULL;
int rc = sqlite3_prepare_v2(g_db, sql, -1, &raw_stmt, NULL);
if (rc != SQLITE_OK) return rc;
db_stmt_t* wrapper = (db_stmt_t*)malloc(sizeof(db_stmt_t));
if (!wrapper) {
sqlite3_finalize(raw_stmt);
return DB_ERROR;
}
wrapper->stmt = raw_stmt;
*out_stmt = wrapper;
return DB_OK;
}
int db_bind_text_param(sqlite3_stmt* stmt, int index, const char* value) {
if (!stmt) return SQLITE_MISUSE;
return sqlite3_bind_text(stmt, index, value ? value : "", -1, SQLITE_TRANSIENT);
int db_bind_text_param(db_stmt_t* stmt, int index, const char* value) {
if (!stmt || !stmt->stmt) return DB_MISUSE;
return sqlite3_bind_text(stmt->stmt, index, value ? value : "", -1, SQLITE_TRANSIENT);
}
int db_bind_int_param(sqlite3_stmt* stmt, int index, int value) {
if (!stmt) return SQLITE_MISUSE;
return sqlite3_bind_int(stmt, index, value);
int db_bind_int_param(db_stmt_t* stmt, int index, int value) {
if (!stmt || !stmt->stmt) return DB_MISUSE;
return sqlite3_bind_int(stmt->stmt, index, value);
}
int db_bind_int64_param(sqlite3_stmt* stmt, int index, long long value) {
if (!stmt) return SQLITE_MISUSE;
return sqlite3_bind_int64(stmt, index, (sqlite3_int64)value);
int db_bind_int64_param(db_stmt_t* stmt, int index, long long value) {
if (!stmt || !stmt->stmt) return DB_MISUSE;
return sqlite3_bind_int64(stmt->stmt, index, (sqlite3_int64)value);
}
int db_step_stmt(sqlite3_stmt* stmt) {
if (!stmt) return SQLITE_MISUSE;
return sqlite3_step(stmt);
int db_step_stmt(db_stmt_t* stmt) {
if (!stmt || !stmt->stmt) return DB_MISUSE;
return sqlite3_step(stmt->stmt);
}
int db_reset_stmt(sqlite3_stmt* stmt) {
if (!stmt) return SQLITE_MISUSE;
return sqlite3_reset(stmt);
int db_reset_stmt(db_stmt_t* stmt) {
if (!stmt || !stmt->stmt) return DB_MISUSE;
return sqlite3_reset(stmt->stmt);
}
const char* db_column_text_value(sqlite3_stmt* stmt, int col) {
if (!stmt) return NULL;
return (const char*)sqlite3_column_text(stmt, col);
const char* db_column_text_value(db_stmt_t* stmt, int col) {
if (!stmt || !stmt->stmt) return NULL;
return (const char*)sqlite3_column_text(stmt->stmt, col);
}
int db_column_int_value(sqlite3_stmt* stmt, int col) {
if (!stmt) return 0;
return sqlite3_column_int(stmt, col);
int db_column_int_value(db_stmt_t* stmt, int col) {
if (!stmt || !stmt->stmt) return 0;
return sqlite3_column_int(stmt->stmt, col);
}
long long db_column_int64_value(sqlite3_stmt* stmt, int col) {
if (!stmt) return 0;
return (long long)sqlite3_column_int64(stmt, col);
long long db_column_int64_value(db_stmt_t* stmt, int col) {
if (!stmt || !stmt->stmt) return 0;
return (long long)sqlite3_column_int64(stmt->stmt, col);
}
double db_column_double_value(sqlite3_stmt* stmt, int col) {
if (!stmt) return 0.0;
return sqlite3_column_double(stmt, col);
double db_column_double_value(db_stmt_t* stmt, int col) {
if (!stmt || !stmt->stmt) return 0.0;
return sqlite3_column_double(stmt->stmt, col);
}
void db_finalize_stmt(sqlite3_stmt* stmt) {
if (stmt) sqlite3_finalize(stmt);
void db_finalize_stmt(db_stmt_t* stmt) {
if (!stmt) return;
if (stmt->stmt) sqlite3_finalize(stmt->stmt);
free(stmt);
}
int db_log_subscription_created(const char* sub_id, const char* wsi_ptr,
src/db_ops.h
+21 -13
View File
@@ -3,27 +3,35 @@
#include <stddef.h>
#include <time.h>
#include <sqlite3.h>
#include <cjson/cJSON.h>
// Generic helpers
int db_is_available(void);
sqlite3* db_get_handle(void);
const char* db_last_error(void);
const char* db_get_database_path(void);
// DB result codes (backend-agnostic)
#define DB_OK 0
#define DB_ERROR 1
#define DB_CONSTRAINT 19
#define DB_MISUSE 21
#define DB_ROW 100
#define DB_DONE 101
typedef struct db_stmt db_stmt_t;
// Generic statement helpers (Phase 1 migration)
int db_prepare(const char* sql, sqlite3_stmt** out_stmt);
int db_bind_text_param(sqlite3_stmt* stmt, int index, const char* value);
int db_bind_int_param(sqlite3_stmt* stmt, int index, int value);
int db_bind_int64_param(sqlite3_stmt* stmt, int index, long long value);
int db_step_stmt(sqlite3_stmt* stmt);
int db_reset_stmt(sqlite3_stmt* stmt);
const char* db_column_text_value(sqlite3_stmt* stmt, int col);
int db_column_int_value(sqlite3_stmt* stmt, int col);
long long db_column_int64_value(sqlite3_stmt* stmt, int col);
double db_column_double_value(sqlite3_stmt* stmt, int col);
void db_finalize_stmt(sqlite3_stmt* stmt);
int db_prepare(const char* sql, db_stmt_t** out_stmt);
int db_bind_text_param(db_stmt_t* stmt, int index, const char* value);
int db_bind_int_param(db_stmt_t* stmt, int index, int value);
int db_bind_int64_param(db_stmt_t* stmt, int index, long long value);
int db_step_stmt(db_stmt_t* stmt);
int db_reset_stmt(db_stmt_t* stmt);
const char* db_column_text_value(db_stmt_t* stmt, int col);
int db_column_int_value(db_stmt_t* stmt, int col);
long long db_column_int64_value(db_stmt_t* stmt, int col);
double db_column_double_value(db_stmt_t* stmt, int col);
void db_finalize_stmt(db_stmt_t* stmt);
// Subscription logging
int db_log_subscription_created(const char* sub_id, const char* wsi_ptr,
src/ip_ban.c
+49 -53
View File
@@ -102,8 +102,7 @@ void ip_ban_init(void) {
}
void ip_ban_load_from_db(void) {
sqlite3* db = db_get_handle();
if (!db || !g_initialized) return;
if (!db_is_available() || !g_initialized) return;
// Create table if it doesn't exist (handles existing databases)
const char* create_sql =
@@ -121,18 +120,16 @@ void ip_ban_load_from_db(void) {
" first_seen INTEGER NOT NULL DEFAULT 0,"
" updated_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))"
");";
char* err = NULL;
if (sqlite3_exec(db, create_sql, NULL, NULL, &err) != SQLITE_OK) {
DEBUG_ERROR("Failed to create ip_bans table: %s", err ? err : "unknown");
if (err) sqlite3_free(err);
if (db_exec_sql(create_sql) != 0) {
DEBUG_ERROR("Failed to create ip_bans table: %s", db_last_error());
return;
}
// Migration: Add idle_* columns if they don't exist (ignore errors if already exists)
sqlite3_exec(db, "ALTER TABLE ip_bans ADD COLUMN idle_failure_count INTEGER NOT NULL DEFAULT 0", NULL, NULL, NULL);
sqlite3_exec(db, "ALTER TABLE ip_bans ADD COLUMN idle_ban_count INTEGER NOT NULL DEFAULT 0", NULL, NULL, NULL);
sqlite3_exec(db, "ALTER TABLE ip_bans ADD COLUMN idle_banned_until INTEGER NOT NULL DEFAULT 0", NULL, NULL, NULL);
sqlite3_exec(db, "ALTER TABLE ip_bans ADD COLUMN idle_first_failure INTEGER NOT NULL DEFAULT 0", NULL, NULL, NULL);
(void)db_exec_sql("ALTER TABLE ip_bans ADD COLUMN idle_failure_count INTEGER NOT NULL DEFAULT 0");
(void)db_exec_sql("ALTER TABLE ip_bans ADD COLUMN idle_ban_count INTEGER NOT NULL DEFAULT 0");
(void)db_exec_sql("ALTER TABLE ip_bans ADD COLUMN idle_banned_until INTEGER NOT NULL DEFAULT 0");
(void)db_exec_sql("ALTER TABLE ip_bans ADD COLUMN idle_first_failure INTEGER NOT NULL DEFAULT 0");
const char* sql =
"SELECT ip, failure_count, ban_count, banned_until, first_failure,"
@@ -141,16 +138,16 @@ void ip_ban_load_from_db(void) {
" idle_failure_count, idle_ban_count, idle_banned_until, idle_first_failure"
" FROM ip_bans";
sqlite3_stmt* stmt;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) {
db_stmt_t* stmt;
if (db_prepare(sql, &stmt) != DB_OK) {
DEBUG_ERROR("Failed to prepare ip_bans load query");
return;
}
int loaded = 0;
pthread_mutex_lock(&g_ban_mutex);
while (sqlite3_step(stmt) == SQLITE_ROW) {
const char* ip = (const char*)sqlite3_column_text(stmt, 0);
while (db_step_stmt(stmt) == DB_ROW) {
const char* ip = db_column_text_value(stmt, 0);
if (!ip) continue;
int idx = find_slot(ip);
@@ -160,25 +157,25 @@ void ip_ban_load_from_db(void) {
entry->state = IP_BAN_ACTIVE;
strncpy(entry->ip, ip, sizeof(entry->ip) - 1);
entry->ip[sizeof(entry->ip) - 1] = '\0';
entry->failure_count = sqlite3_column_int(stmt, 1);
entry->ban_count = sqlite3_column_int(stmt, 2);
entry->banned_until = (time_t)sqlite3_column_int64(stmt, 3);
entry->first_failure = (time_t)sqlite3_column_int64(stmt, 4);
entry->has_authed_successfully = sqlite3_column_int(stmt, 5);
entry->last_success_at = (time_t)sqlite3_column_int64(stmt, 6);
entry->total_connections = sqlite3_column_int(stmt, 7);
entry->total_failures = sqlite3_column_int(stmt, 8);
entry->total_successes = sqlite3_column_int(stmt, 9);
entry->first_seen = (time_t)sqlite3_column_int64(stmt, 10);
entry->failure_count = db_column_int_value(stmt, 1);
entry->ban_count = db_column_int_value(stmt, 2);
entry->banned_until = (time_t)db_column_int64_value(stmt, 3);
entry->first_failure = (time_t)db_column_int64_value(stmt, 4);
entry->has_authed_successfully = db_column_int_value(stmt, 5);
entry->last_success_at = (time_t)db_column_int64_value(stmt, 6);
entry->total_connections = db_column_int_value(stmt, 7);
entry->total_failures = db_column_int_value(stmt, 8);
entry->total_successes = db_column_int_value(stmt, 9);
entry->first_seen = (time_t)db_column_int64_value(stmt, 10);
// Load idle tracking fields (default to 0 if columns don't exist yet)
entry->idle_failure_count = sqlite3_column_int(stmt, 11);
entry->idle_ban_count = sqlite3_column_int(stmt, 12);
entry->idle_banned_until = (time_t)sqlite3_column_int64(stmt, 13);
entry->idle_first_failure = (time_t)sqlite3_column_int64(stmt, 14);
entry->idle_failure_count = db_column_int_value(stmt, 11);
entry->idle_ban_count = db_column_int_value(stmt, 12);
entry->idle_banned_until = (time_t)db_column_int64_value(stmt, 13);
entry->idle_first_failure = (time_t)db_column_int64_value(stmt, 14);
loaded++;
}
pthread_mutex_unlock(&g_ban_mutex);
sqlite3_finalize(stmt);
db_finalize_stmt(stmt);
// Count how many are still actively banned
time_t now = time(NULL);
@@ -196,8 +193,7 @@ void ip_ban_load_from_db(void) {
}
void ip_ban_save_to_db(void) {
sqlite3* db = db_get_handle();
if (!db || !g_initialized) return;
if (!db_is_available() || !g_initialized) return;
const char* upsert_sql =
"INSERT OR REPLACE INTO ip_bans"
@@ -207,13 +203,13 @@ void ip_ban_save_to_db(void) {
" idle_failure_count, idle_ban_count, idle_banned_until, idle_first_failure)"
" VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, strftime('%s', 'now'), ?, ?, ?, ?)";
sqlite3_stmt* stmt;
if (sqlite3_prepare_v2(db, upsert_sql, -1, &stmt, NULL) != SQLITE_OK) {
db_stmt_t* stmt;
if (db_prepare(upsert_sql, &stmt) != DB_OK) {
DEBUG_ERROR("Failed to prepare ip_bans save query");
return;
}
sqlite3_exec(db, "BEGIN TRANSACTION", NULL, NULL, NULL);
(void)db_exec_sql("BEGIN TRANSACTION");
int saved = 0;
pthread_mutex_lock(&g_ban_mutex);
@@ -221,25 +217,25 @@ void ip_ban_save_to_db(void) {
ip_ban_entry_t* entry = &g_ban_table[i];
if (entry->state != IP_BAN_ACTIVE) continue;
sqlite3_reset(stmt);
sqlite3_bind_text(stmt, 1, entry->ip, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, entry->failure_count);
sqlite3_bind_int(stmt, 3, entry->ban_count);
sqlite3_bind_int64(stmt, 4, (sqlite3_int64)entry->banned_until);
sqlite3_bind_int64(stmt, 5, (sqlite3_int64)entry->first_failure);
sqlite3_bind_int(stmt, 6, entry->has_authed_successfully);
sqlite3_bind_int64(stmt, 7, (sqlite3_int64)entry->last_success_at);
sqlite3_bind_int(stmt, 8, entry->total_connections);
sqlite3_bind_int(stmt, 9, entry->total_failures);
sqlite3_bind_int(stmt, 10, entry->total_successes);
sqlite3_bind_int64(stmt, 11, (sqlite3_int64)entry->first_seen);
(void)db_reset_stmt(stmt);
db_bind_text_param(stmt, 1, entry->ip);
db_bind_int_param(stmt, 2, entry->failure_count);
db_bind_int_param(stmt, 3, entry->ban_count);
db_bind_int64_param(stmt, 4, (long long)entry->banned_until);
db_bind_int64_param(stmt, 5, (long long)entry->first_failure);
db_bind_int_param(stmt, 6, entry->has_authed_successfully);
db_bind_int64_param(stmt, 7, (long long)entry->last_success_at);
db_bind_int_param(stmt, 8, entry->total_connections);
db_bind_int_param(stmt, 9, entry->total_failures);
db_bind_int_param(stmt, 10, entry->total_successes);
db_bind_int64_param(stmt, 11, (long long)entry->first_seen);
// Idle tracking fields
sqlite3_bind_int(stmt, 12, entry->idle_failure_count);
sqlite3_bind_int(stmt, 13, entry->idle_ban_count);
sqlite3_bind_int64(stmt, 14, (sqlite3_int64)entry->idle_banned_until);
sqlite3_bind_int64(stmt, 15, (sqlite3_int64)entry->idle_first_failure);
db_bind_int_param(stmt, 12, entry->idle_failure_count);
db_bind_int_param(stmt, 13, entry->idle_ban_count);
db_bind_int64_param(stmt, 14, (long long)entry->idle_banned_until);
db_bind_int64_param(stmt, 15, (long long)entry->idle_first_failure);
if (sqlite3_step(stmt) != SQLITE_DONE) {
if (db_step_stmt(stmt) != DB_DONE) {
DEBUG_WARN("Failed to save ip_ban entry for %s", entry->ip);
} else {
saved++;
@@ -247,8 +243,8 @@ void ip_ban_save_to_db(void) {
}
pthread_mutex_unlock(&g_ban_mutex);
sqlite3_finalize(stmt);
sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
db_finalize_stmt(stmt);
(void)db_exec_sql("COMMIT");
DEBUG_TRACE("IP ban table saved: %d entries written to DB", saved);
}
src/main.c
+9 -9
View File
@@ -426,7 +426,7 @@ int init_database(const char* database_path_override) {
cleanup_stale_wal_files(db_path);
int rc = sqlite3_open(db_path, &g_db);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
DEBUG_ERROR("Cannot open database");
DEBUG_TRACE("Exiting init_database() - failed to open database");
return -1;
@@ -730,7 +730,7 @@ int store_event(cJSON* event) {
return -1;
}
int rc = SQLITE_ERROR;
int rc = DB_ERROR;
int extended_errcode = 0;
if (db_insert_event_with_json(cJSON_GetStringValue(id),
cJSON_GetStringValue(pubkey),
@@ -749,15 +749,15 @@ int store_event(cJSON* event) {
return -1;
}
if (rc != SQLITE_DONE) {
if (rc != DB_DONE) {
const char* err_msg = db_last_error();
if (rc != SQLITE_CONSTRAINT) {
if (rc != DB_CONSTRAINT) {
DEBUG_ERROR("INSERT failed: rc=%d, extended_errcode=%d, msg=%s", rc, extended_errcode, err_msg);
}
}
if (rc != SQLITE_DONE) {
if (rc == SQLITE_CONSTRAINT) {
if (rc != DB_DONE) {
if (rc == DB_CONSTRAINT) {
DEBUG_WARN("Event already exists in database");
// Add TRACE level debug to show both events
@@ -1359,9 +1359,9 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
clock_gettime(CLOCK_MONOTONIC, &query_start);
// Execute query and send events
sqlite3_stmt* stmt;
db_stmt_t* stmt;
int rc = db_prepare(sql, &stmt);
if (rc != SQLITE_OK) {
if (rc != DB_OK) {
char error_msg[256];
snprintf(error_msg, sizeof(error_msg), "Failed to prepare subscription query: %s", db_last_error());
DEBUG_ERROR(error_msg);
@@ -1399,7 +1399,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
int filter_responses = get_config_bool("expiration_filter", 1);
int row_count = 0;
while (db_step_stmt(stmt) == SQLITE_ROW) {
while (db_step_stmt(stmt) == DB_ROW) {
row_count++;
// Track rows returned for abuse detection
src/main.h
+2 -2
View File
@@ -13,8 +13,8 @@
// Using CRELAY_ prefix to avoid conflicts with nostr_core_lib VERSION macros
#define CRELAY_VERSION_MAJOR 2
#define CRELAY_VERSION_MINOR 0
#define CRELAY_VERSION_PATCH 3
#define CRELAY_VERSION "v2.0.3"
#define CRELAY_VERSION_PATCH 4
#define CRELAY_VERSION "v2.0.4"
// Relay metadata (authoritative source for NIP-11 information)
#define RELAY_NAME "C-Relay"