#define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include // Include nostr_core_lib for Nostr functionality #include "../nostr_core_lib/cjson/cJSON.h" #include "../nostr_core_lib/nostr_core/nostr_core.h" #include "../nostr_core_lib/nostr_core/nip013.h" // NIP-13: Proof of Work #include "../nostr_core_lib/nostr_core/nip019.h" // NIP-19: bech32-encoded entities #include "main.h" // Version and relay metadata #include "config.h" // Configuration management system #include "sql_schema.h" // Embedded database schema #include "websockets.h" // WebSocket protocol implementation #include "subscriptions.h" // Subscription management system #include "debug.h" // Debug system #include "thread_pool.h" // Thread pool scaffold #include "db_ops.h" // Forward declarations for unified request validator int nostr_validate_unified_request(const char* json_string, size_t json_length); int ginxsom_request_validator_init(const char* db_path, const char* app_name); void ginxsom_request_validator_cleanup(void); // Forward declarations for NIP-42 functions from request_validator.c int nostr_nip42_generate_challenge(char *challenge_buffer, size_t buffer_size); int nostr_nip42_verify_auth_event(cJSON *event, const char *challenge_id, const char *relay_url, int time_tolerance_seconds); // Color constants for logging #define RED "\033[31m" #define GREEN "\033[32m" #define YELLOW "\033[33m" #define BLUE "\033[34m" #define BOLD "\033[1m" #define RESET "\033[0m" // Global state sqlite3* g_db = NULL; // Non-static so config.c can access it int g_server_running = 1; // Non-static so websockets.c can access it volatile sig_atomic_t g_shutdown_flag = 0; // Non-static so config.c can access it for restart functionality int g_restart_requested = 0; // Non-static so config.c can access it for restart functionality struct lws_context *ws_context = NULL; // Non-static so websockets.c can access it // NIP-11 relay information structure struct relay_info { char name[RELAY_NAME_MAX_LENGTH]; char description[RELAY_DESCRIPTION_MAX_LENGTH]; char banner[RELAY_URL_MAX_LENGTH]; char icon[RELAY_URL_MAX_LENGTH]; char pubkey[RELAY_PUBKEY_MAX_LENGTH]; char contact[RELAY_CONTACT_MAX_LENGTH]; char software[RELAY_URL_MAX_LENGTH]; char version[64]; char privacy_policy[RELAY_URL_MAX_LENGTH]; char terms_of_service[RELAY_URL_MAX_LENGTH]; cJSON* supported_nips; // Array of supported NIP numbers cJSON* limitation; // Server limitations object cJSON* retention; // Event retention policies array cJSON* relay_countries; // Array of country codes cJSON* language_tags; // Array of language tags cJSON* tags; // Array of content tags char posting_policy[RELAY_URL_MAX_LENGTH]; cJSON* fees; // Payment fee structure char payments_url[RELAY_URL_MAX_LENGTH]; }; // NIP-40 Expiration configuration (now in nip040.c) extern struct expiration_config g_expiration_config; // Global subscription manager instance (defined in websockets.c) extern subscription_manager_t g_subscription_manager; ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // DATA STRUCTURES ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Forward declaration for subscription manager configuration void update_subscription_manager_config(void); // Forward declarations for subscription database logging void log_subscription_created(const subscription_t* sub); void log_subscription_closed(const char* sub_id, const char* client_ip, const char* reason); void log_subscription_disconnected(const char* client_ip); void update_subscription_events_sent(const char* sub_id, int events_sent); // Forward declarations for NIP-01 event handling const char* extract_d_tag_value(cJSON* tags); int check_and_handle_replaceable_event(int kind, const char* pubkey, long created_at); int check_and_handle_addressable_event(int kind, const char* pubkey, const char* d_tag_value, long created_at); int handle_event_message(cJSON* event, char* error_message, size_t error_size); // Forward declaration for unified validation int nostr_validate_unified_request(const char* json_string, size_t json_length); // Forward declaration for admin event processing (kind 23456) int process_admin_event_in_config(cJSON* event, char* error_message, size_t error_size, struct lws* wsi); // Forward declaration for NIP-45 COUNT message handling int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss); // Forward declaration for NOTICE message support void send_notice_message(struct lws* wsi, struct per_session_data* pss, const char* message); // Thread pool wake callback (called by worker threads) static void wake_event_loop_from_thread_pool(void* ctx); // Parameter binding helpers for SQL queries static void add_bind_param(char*** params, int* count, int* capacity, const char* value) { if (*count >= *capacity) { *capacity = *capacity == 0 ? 16 : *capacity * 2; *params = realloc(*params, *capacity * sizeof(char*)); } (*params)[(*count)++] = strdup(value); } static void free_bind_params(char** params, int count) { for (int i = 0; i < count; i++) { free(params[i]); } free(params); } typedef struct req_async_state { char sub_id[SUBSCRIPTION_ID_MAX_LENGTH]; struct lws* wsi_token; pthread_mutex_t mutex; int pending_jobs; } req_async_state_t; typedef struct req_async_submit_ctx { req_async_state_t* state; } req_async_submit_ctx_t; typedef struct req_async_completion { req_async_state_t* state; thread_pool_status_t status; thread_pool_req_result_t* req_result; struct req_async_completion* next; } req_async_completion_t; static pthread_mutex_t g_req_async_completion_mutex = PTHREAD_MUTEX_INITIALIZER; static req_async_completion_t* g_req_async_completion_head = NULL; static req_async_completion_t* g_req_async_completion_tail = NULL; static void free_req_payload_main(void* p) { thread_pool_req_payload_t* payload = (thread_pool_req_payload_t*)p; if (!payload) return; free(payload->sql); if (payload->bind_params) { for (int i = 0; i < payload->bind_param_count; i++) { free(payload->bind_params[i]); } free(payload->bind_params); } free(payload); } static req_async_state_t* req_async_state_create(const char* sub_id, struct lws* wsi) { req_async_state_t* state = calloc(1, sizeof(*state)); if (!state) { return NULL; } if (sub_id) { strncpy(state->sub_id, sub_id, sizeof(state->sub_id) - 1); state->sub_id[sizeof(state->sub_id) - 1] = '\0'; } state->wsi_token = wsi; pthread_mutex_init(&state->mutex, NULL); return state; } static void req_async_state_free(req_async_state_t* state) { if (!state) return; pthread_mutex_destroy(&state->mutex); free(state); } static void req_async_state_increment_pending(req_async_state_t* state) { if (!state) return; pthread_mutex_lock(&state->mutex); state->pending_jobs++; pthread_mutex_unlock(&state->mutex); } static int req_async_state_decrement_pending(req_async_state_t* state) { if (!state) return 0; pthread_mutex_lock(&state->mutex); if (state->pending_jobs > 0) { state->pending_jobs--; } int remaining = state->pending_jobs; pthread_mutex_unlock(&state->mutex); return remaining; } static void req_async_completion_push(req_async_completion_t* completion) { if (!completion) return; pthread_mutex_lock(&g_req_async_completion_mutex); completion->next = NULL; if (!g_req_async_completion_tail) { g_req_async_completion_head = completion; g_req_async_completion_tail = completion; } else { g_req_async_completion_tail->next = completion; g_req_async_completion_tail = completion; } pthread_mutex_unlock(&g_req_async_completion_mutex); } static req_async_completion_t* req_async_completion_pop(void) { pthread_mutex_lock(&g_req_async_completion_mutex); req_async_completion_t* completion = g_req_async_completion_head; if (completion) { g_req_async_completion_head = completion->next; if (!g_req_async_completion_head) { g_req_async_completion_tail = NULL; } } pthread_mutex_unlock(&g_req_async_completion_mutex); return completion; } static int resolve_req_async_target(req_async_state_t* state, struct lws** out_wsi, struct per_session_data** out_pss) { if (!state || !out_wsi || !out_pss) { return 0; } *out_wsi = NULL; *out_pss = NULL; pthread_mutex_lock(&g_subscription_manager.subscriptions_lock); subscription_t* sub = g_subscription_manager.active_subscriptions; while (sub) { if (sub->active && sub->wsi == state->wsi_token && strcmp(sub->id, state->sub_id) == 0) { *out_wsi = sub->wsi; break; } sub = sub->next; } pthread_mutex_unlock(&g_subscription_manager.subscriptions_lock); if (!*out_wsi) { return 0; } *out_pss = (struct per_session_data*)lws_wsi_user(*out_wsi); return (*out_pss != NULL); } static void send_eose_message(struct lws* wsi, struct per_session_data* pss, const char* sub_id) { if (!wsi || !pss || !sub_id) { return; } cJSON* eose_response = cJSON_CreateArray(); if (!eose_response) { return; } cJSON_AddItemToArray(eose_response, cJSON_CreateString("EOSE")); cJSON_AddItemToArray(eose_response, cJSON_CreateString(sub_id)); char *eose_str = cJSON_Print(eose_response); if (eose_str) { size_t eose_len = strlen(eose_str); DEBUG_TRACE("WS_FRAME_SEND: type=EOSE len=%zu data=%.100s%s", eose_len, eose_str, eose_len > 100 ? "..." : ""); if (queue_message(wsi, pss, eose_str, eose_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue EOSE message"); } free(eose_str); } cJSON_Delete(eose_response); } static void req_async_result_cb(const thread_pool_result_t* result, void* user_ctx) { req_async_submit_ctx_t* ctx = (req_async_submit_ctx_t*)user_ctx; if (!ctx || !ctx->state || !result) { if (result && result->result_data) { thread_pool_free_req_result((thread_pool_req_result_t*)result->result_data); } if (ctx) free(ctx); return; } req_async_completion_t* completion = calloc(1, sizeof(*completion)); if (!completion) { if (result->result_data) { thread_pool_free_req_result((thread_pool_req_result_t*)result->result_data); } free(ctx); return; } completion->state = ctx->state; completion->status = result->status; completion->req_result = (thread_pool_req_result_t*)result->result_data; req_async_completion_push(completion); wake_event_loop_from_thread_pool(NULL); free(ctx); } static int submit_req_query_async(req_async_state_t* state, const char* sql, const char** bind_params, int bind_param_count) { if (!state || !sql) { return -1; } thread_pool_req_payload_t* payload = calloc(1, sizeof(*payload)); if (!payload) { return -1; } payload->sql = strdup(sql); payload->bind_param_count = bind_param_count; if (!payload->sql) { free_req_payload_main(payload); return -1; } if (bind_param_count > 0) { payload->bind_params = calloc((size_t)bind_param_count, sizeof(char*)); if (!payload->bind_params) { free_req_payload_main(payload); return -1; } for (int i = 0; i < bind_param_count; i++) { const char* v = (bind_params && bind_params[i]) ? bind_params[i] : ""; payload->bind_params[i] = strdup(v); if (!payload->bind_params[i]) { free_req_payload_main(payload); return -1; } } } req_async_submit_ctx_t* ctx = calloc(1, sizeof(*ctx)); if (!ctx) { free_req_payload_main(payload); return -1; } ctx->state = state; thread_pool_job_t job; memset(&job, 0, sizeof(job)); job.type = THREAD_POOL_JOB_REQ_QUERY; job.payload = payload; job.payload_size = sizeof(*payload); job.payload_free = free_req_payload_main; job.result_cb = req_async_result_cb; job.result_cb_ctx = ctx; req_async_state_increment_pending(state); thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL); if (submit_rc != THREAD_POOL_STATUS_OK) { req_async_state_decrement_pending(state); free(ctx); free_req_payload_main(payload); return -1; } return 0; } void process_req_async_completions(void) { req_async_completion_t* completion = NULL; while ((completion = req_async_completion_pop()) != NULL) { struct lws* target_wsi = NULL; struct per_session_data* target_pss = NULL; int has_target = resolve_req_async_target(completion->state, &target_wsi, &target_pss); if (has_target && completion->status == THREAD_POOL_STATUS_OK && completion->req_result) { for (int r = 0; r < completion->req_result->row_count; r++) { const char* event_json_str = completion->req_result->event_json_rows[r]; if (!event_json_str) { continue; } size_t sub_id_len = strlen(completion->state->sub_id); size_t event_json_len = strlen(event_json_str); size_t msg_len = 10 + sub_id_len + 3 + event_json_len + 1; unsigned char* buf = malloc(LWS_PRE + msg_len + 1); if (!buf) { continue; } char* msg_ptr = (char*)(buf + LWS_PRE); snprintf(msg_ptr, msg_len + 1, "[\"EVENT\",\"%s\",%s]", completion->state->sub_id, event_json_str); size_t actual_len = strlen(msg_ptr); if (queue_message_take_ownership(target_wsi, target_pss, buf, actual_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue async EVENT message for sub=%s", completion->state->sub_id); } } } else if (has_target && completion->status != THREAD_POOL_STATUS_OK) { send_notice_message(target_wsi, target_pss, "error: failed to execute subscription query"); } int remaining = req_async_state_decrement_pending(completion->state); if (remaining == 0) { if (has_target) { send_eose_message(target_wsi, target_pss, completion->state->sub_id); } req_async_state_free(completion->state); } if (completion->req_result) { thread_pool_free_req_result(completion->req_result); } free(completion); } } // Forward declaration for enhanced admin event authorization int is_authorized_admin_event(cJSON* event, char* error_message, size_t error_size); // Forward declarations for NIP-42 authentication functions void send_nip42_auth_challenge(struct lws* wsi, struct per_session_data* pss); void handle_nip42_auth_signed_event(struct lws* wsi, struct per_session_data* pss, cJSON* auth_event); void handle_nip42_auth_challenge_response(struct lws* wsi, struct per_session_data* pss, const char* challenge); // Forward declarations for NIP-09 deletion request handling int handle_deletion_request(cJSON* event, char* error_message, size_t error_size); 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); int mark_event_as_deleted(const char* event_id, const char* deletion_event_id, const char* reason); // Forward declaration for database functions int store_event(cJSON* event); cJSON* retrieve_event(const char* event_id); // Forward declaration for monitoring system void monitoring_on_event_stored(void); // Forward declarations for NIP-11 relay information handling void init_relay_info(); void cleanup_relay_info(); cJSON* generate_relay_info_json(); 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); // Forward declarations for IP ban system void ip_ban_init(void); void ip_ban_load_from_db(void); // Forward declarations for NIP-13 PoW handling (now in nip013.c) void init_pow_config(); int validate_event_pow(cJSON* event, char* error_message, size_t error_size); // Forward declarations for NIP-40 expiration handling (now in nip040.c) void init_expiration_config(); long extract_expiration_timestamp(cJSON* tags); int is_event_expired(cJSON* event, time_t current_time); int validate_event_expiration(cJSON* event, char* error_message, size_t error_size); ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // LOGGING FUNCTIONS ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Logging functions - REMOVED (replaced by debug system in debug.c) static void nostr_log_cb(int level, const char* component, const char* message, void* user_data) { (void)user_data; const char* comp = (component && component[0]) ? component : "core"; const char* msg = message ? message : ""; switch (level) { case NOSTR_LOG_LEVEL_ERROR: DEBUG_ERROR("[nostr:%s] %s", comp, msg); break; case NOSTR_LOG_LEVEL_WARN: DEBUG_WARN("[nostr:%s] %s", comp, msg); break; case NOSTR_LOG_LEVEL_INFO: DEBUG_INFO("[nostr:%s] %s", comp, msg); break; case NOSTR_LOG_LEVEL_DEBUG: DEBUG_LOG("[nostr:%s] %s", comp, msg); break; case NOSTR_LOG_LEVEL_TRACE: default: DEBUG_TRACE("[nostr:%s] %s", comp, msg); break; } } static nostr_log_level_t nostr_log_level_from_debug_level(int debug_level) { if (debug_level >= DEBUG_LEVEL_TRACE) { return NOSTR_LOG_LEVEL_TRACE; } if (debug_level >= DEBUG_LEVEL_DEBUG) { return NOSTR_LOG_LEVEL_DEBUG; } if (debug_level >= DEBUG_LEVEL_INFO) { return NOSTR_LOG_LEVEL_INFO; } if (debug_level >= DEBUG_LEVEL_WARN) { return NOSTR_LOG_LEVEL_WARN; } if (debug_level >= DEBUG_LEVEL_ERROR) { return NOSTR_LOG_LEVEL_ERROR; } // Production-safe default for systemd/journald deployments. return NOSTR_LOG_LEVEL_INFO; } // Update subscription manager configuration from config system void update_subscription_manager_config(void) { g_subscription_manager.max_subscriptions_per_client = get_config_int("max_subscriptions_per_client", MAX_SUBSCRIPTIONS_PER_CLIENT); g_subscription_manager.max_total_subscriptions = get_config_int("max_total_subscriptions", MAX_TOTAL_SUBSCRIPTIONS); char config_msg[256]; snprintf(config_msg, sizeof(config_msg), "Subscription limits: max_per_client=%d, max_total=%d", g_subscription_manager.max_subscriptions_per_client, g_subscription_manager.max_total_subscriptions); } // Signal handler for graceful shutdown void signal_handler(int sig) { if (sig == SIGINT || sig == SIGTERM) { g_server_running = 0; } } static void wake_event_loop_from_thread_pool(void* ctx) { (void)ctx; if (ws_context) { lws_cancel_service(ws_context); } } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // NOTICE MESSAGE SUPPORT ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Send NOTICE message to client (NIP-01) void send_notice_message(struct lws* wsi, struct per_session_data* pss, const char* message) { if (!wsi || !message) return; cJSON* notice_msg = cJSON_CreateArray(); cJSON_AddItemToArray(notice_msg, cJSON_CreateString("NOTICE")); cJSON_AddItemToArray(notice_msg, cJSON_CreateString(message)); char* msg_str = cJSON_Print(notice_msg); if (msg_str) { size_t msg_len = strlen(msg_str); // Use proper message queue system instead of direct lws_write if (queue_message(wsi, pss, msg_str, msg_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue NOTICE message"); } free(msg_str); } cJSON_Delete(notice_msg); } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // DATABASE QUERY LOGGING ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// /** * Log database query execution with timing and context * Only logs at debug level 3 (DEBUG) or higher * Warns if query takes >10ms (slow query) * * @param query_type Type of query (REQ, COUNT, INSERT, CONFIG, etc.) * @param sub_id Subscription ID (NULL if not applicable) * @param client_ip Client IP address (NULL if not applicable) * @param sql SQL query text * @param elapsed_us Execution time in microseconds * @param rows_returned Number of rows returned or affected */ void log_query_execution(const char* query_type, const char* sub_id, const char* client_ip, const char* sql, long elapsed_us, int rows_returned) { // Only log at debug level 3 (INFO) or higher if (g_debug_level < DEBUG_LEVEL_INFO) { return; } // Truncate SQL if too long (keep first 500 chars) char sql_truncated[512]; if (strlen(sql) > 500) { snprintf(sql_truncated, sizeof(sql_truncated), "%.497s...", sql); } else { snprintf(sql_truncated, sizeof(sql_truncated), "%s", sql); } // Get timestamp time_t now = time(NULL); struct tm* tm_info = localtime(&now); char timestamp[32]; strftime(timestamp, sizeof(timestamp), "%Y-%m-%d %H:%M:%S", tm_info); // Log query with all context (direct to stdout/stderr, not through DEBUG_LOG) fprintf(stderr, "[%s] [QUERY] type=%s sub=%s ip=%s time=%ldus rows=%d sql=%s\n", timestamp, query_type, sub_id ? sub_id : "N/A", client_ip ? client_ip : "N/A", elapsed_us, rows_returned, sql_truncated); // Warn if query is slow (>10ms = 10000us) if (elapsed_us > 10000) { fprintf(stderr, "[%s] [SLOW_QUERY] %ldms: %s\n", timestamp, elapsed_us / 1000, sql_truncated); } fflush(stderr); } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // DATABASE FUNCTIONS ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Clean up stale SQLite WAL files that may cause lock issues after unclean shutdown static void cleanup_stale_wal_files(const char* db_path) { if (!db_path) return; // Check if database file exists if (access(db_path, F_OK) != 0) { return; // Database doesn't exist yet, nothing to clean } // Build paths for WAL and SHM files char wal_path[1024]; char shm_path[1024]; snprintf(wal_path, sizeof(wal_path), "%s-wal", db_path); snprintf(shm_path, sizeof(shm_path), "%s-shm", db_path); // Check if WAL or SHM files exist int has_wal = (access(wal_path, F_OK) == 0); int has_shm = (access(shm_path, F_OK) == 0); if (has_wal || has_shm) { DEBUG_WARN("Detected stale SQLite WAL files from previous unclean shutdown"); // Try to remove WAL file if (has_wal) { if (unlink(wal_path) == 0) { DEBUG_INFO("Removed stale WAL file"); } else { char error_msg[256]; snprintf(error_msg, sizeof(error_msg), "Failed to remove WAL file: %s", strerror(errno)); DEBUG_WARN(error_msg); } } // Try to remove SHM file if (has_shm) { if (unlink(shm_path) == 0) { DEBUG_INFO("Removed stale SHM file"); } else { char error_msg[256]; snprintf(error_msg, sizeof(error_msg), "Failed to remove SHM file: %s", strerror(errno)); DEBUG_WARN(error_msg); } } } } // Initialize database connection and schema int init_database(const char* database_path_override) { DEBUG_TRACE("Entering init_database()"); // Priority 1: Command line database path override const char* db_path = database_path_override; // Priority 2: Configuration system (if available) if (!db_path) { db_path = get_config_value("database_path"); } // Priority 3: Default path if (!db_path) { db_path = DEFAULT_DATABASE_PATH; } DEBUG_LOG("Initializing database: %s", db_path); // Clean up stale WAL files before opening database cleanup_stale_wal_files(db_path); int rc = db_init(db_path); if (rc != DB_OK) { DEBUG_ERROR("Cannot open database"); DEBUG_TRACE("Exiting init_database() - failed to open database"); return -1; } // DEBUG_GUARD_START if (g_debug_level >= DEBUG_LEVEL_DEBUG) { // Check config table row count immediately after database open int row_count = 0; if (db_get_config_row_count(&row_count) == 0) { DEBUG_LOG("Config table row count immediately after db_init(): %d", row_count); } else { DEBUG_LOG("Config table count unavailable immediately after sqlite3_open() (table may not exist yet)"); } } // DEBUG_GUARD_END // Check if database is already initialized by looking for the events table int has_events_table = 0; if (db_table_exists("events", &has_events_table) == 0) { if (has_events_table) { // Check existing schema version and migrate if needed char* db_version = db_get_schema_version_dup(); int needs_migration = 0; // Check if migration is needed if (!db_version || strcmp(db_version, "5") == 0) { needs_migration = 1; } else if (strcmp(db_version, "6") == 0) { // Database is at schema version v6 (compatible) } else if (strcmp(db_version, "7") == 0) { // Database is at schema version v7 (compatible) } else if (strcmp(db_version, "8") == 0) { // Database is at schema version v8 (compatible) } else if (strcmp(db_version, "9") == 0) { // Database is at schema version v9 (compatible) } else if (strcmp(db_version, "10") == 0) { // Database is at schema version v10 (compatible) } else if (strcmp(db_version, EMBEDDED_SCHEMA_VERSION) == 0) { // Database is at current schema version } else { char warning_msg[256]; snprintf(warning_msg, sizeof(warning_msg), "Unknown database schema version: %s (expected %s)", db_version, EMBEDDED_SCHEMA_VERSION); DEBUG_WARN(warning_msg); } // Perform migration if needed if (needs_migration) { // Check if auth_rules table already exists int has_auth_rules = 0; (void)db_table_exists("auth_rules", &has_auth_rules); if (!has_auth_rules) { // Add auth_rules table matching sql_schema.h const char* create_auth_rules_sql = "CREATE TABLE IF NOT EXISTS auth_rules (" " id INTEGER PRIMARY KEY AUTOINCREMENT," " rule_type TEXT NOT NULL CHECK (rule_type IN ('whitelist', 'blacklist', 'rate_limit', 'auth_required'))," " pattern_type TEXT NOT NULL CHECK (pattern_type IN ('pubkey', 'kind', 'ip', 'global'))," " pattern_value TEXT," " action TEXT NOT NULL CHECK (action IN ('allow', 'deny', 'require_auth', 'rate_limit'))," " parameters TEXT," " active INTEGER NOT NULL DEFAULT 1," " created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))," " updated_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))" ");"; if (db_exec_sql(create_auth_rules_sql) != 0) { DEBUG_ERROR("Failed to create auth_rules table"); if (db_version) free(db_version); return -1; } // Add indexes for auth_rules table const char* create_auth_rules_indexes_sql = "CREATE INDEX IF NOT EXISTS idx_auth_rules_pattern ON auth_rules(pattern_type, pattern_value);" "CREATE INDEX IF NOT EXISTS idx_auth_rules_type ON auth_rules(rule_type);" "CREATE INDEX IF NOT EXISTS idx_auth_rules_active ON auth_rules(active);"; if (db_exec_sql(create_auth_rules_indexes_sql) != 0) { DEBUG_ERROR("Failed to create auth_rules indexes"); if (db_version) free(db_version); return -1; } } else { // auth_rules table already exists, skipping creation } // Update schema version to v6 const char* update_version_sql = "INSERT OR REPLACE INTO schema_info (key, value, updated_at) " "VALUES ('version', '6', strftime('%s', 'now'))"; if (db_exec_sql(update_version_sql) != 0) { DEBUG_ERROR("Failed to update schema version"); if (db_version) free(db_version); return -1; } if (db_version) { free(db_version); } } else if (db_version) { free(db_version); } } else { // Initialize database schema using embedded SQL // Execute the embedded schema SQL if (db_exec_sql(EMBEDDED_SCHEMA_SQL) != 0) { DEBUG_ERROR("Failed to initialize database schema"); return -1; } } } else { DEBUG_ERROR("Failed to check existing database schema"); return -1; } // Enable WAL mode for better concurrency and crash recovery if (db_exec_sql("PRAGMA journal_mode=WAL;") != 0) { DEBUG_WARN("Failed to enable WAL mode"); // Continue anyway - WAL mode is optional } else { DEBUG_LOG("SQLite WAL mode enabled"); } // Apply SQLite performance tuning PRAGMAs from config // mmap_size: memory-map the database file to eliminate pread64 syscall overhead // Default 256MB covers most relay databases; set to 0 to disable long mmap_size = get_config_int("sqlite_mmap_size", 268435456); if (mmap_size > 0) { char mmap_pragma[64]; snprintf(mmap_pragma, sizeof(mmap_pragma), "PRAGMA mmap_size=%ld;", mmap_size); if (db_exec_sql(mmap_pragma) != 0) { DEBUG_WARN("Failed to set mmap_size"); } else { DEBUG_LOG("SQLite mmap_size set to %ld bytes", mmap_size); } } // cache_size_kb: page cache size in KB (negative value = KB, positive = number of 4KB pages) // Default 64MB keeps hot event data in memory and reduces repeated disk reads int cache_size_kb = get_config_int("sqlite_cache_size_kb", 65536); if (cache_size_kb != 0) { char cache_pragma[64]; // Use negative value so SQLite interprets it as KB rather than page count snprintf(cache_pragma, sizeof(cache_pragma), "PRAGMA cache_size=-%d;", cache_size_kb > 0 ? cache_size_kb : -cache_size_kb); if (db_exec_sql(cache_pragma) != 0) { DEBUG_WARN("Failed to set cache_size"); } else { DEBUG_LOG("SQLite cache_size set to %d KB", cache_size_kb); } } DEBUG_TRACE("Exiting init_database() - success"); return 0; } // Close database connection with proper WAL checkpoint void close_database() { DEBUG_TRACE("Entering close_database()"); if (g_db) { // Perform WAL checkpoint to minimize stale files on next startup DEBUG_LOG("Performing WAL checkpoint before database close"); if (db_exec_sql("PRAGMA wal_checkpoint(TRUNCATE);") != 0) { DEBUG_WARN("WAL checkpoint warning"); } db_close(); DEBUG_LOG("Database connection closed"); } DEBUG_TRACE("Exiting close_database()"); } // Event type classification typedef enum { EVENT_TYPE_REGULAR, EVENT_TYPE_REPLACEABLE, EVENT_TYPE_EPHEMERAL, EVENT_TYPE_ADDRESSABLE, EVENT_TYPE_UNKNOWN } event_type_t; event_type_t classify_event_kind(int kind) { if ((kind >= 1000 && kind < 10000) || (kind >= 4 && kind < 45) || kind == 1 || kind == 2) { return EVENT_TYPE_REGULAR; } if ((kind >= 10000 && kind < 20000) || kind == 0 || kind == 3) { return EVENT_TYPE_REPLACEABLE; } if (kind >= 20000 && kind < 30000) { return EVENT_TYPE_EPHEMERAL; } if (kind >= 30000 && kind < 40000) { return EVENT_TYPE_ADDRESSABLE; } return EVENT_TYPE_UNKNOWN; } const char* event_type_to_string(event_type_t type) { switch (type) { case EVENT_TYPE_REGULAR: return "regular"; case EVENT_TYPE_REPLACEABLE: return "replaceable"; case EVENT_TYPE_EPHEMERAL: return "ephemeral"; case EVENT_TYPE_ADDRESSABLE: return "addressable"; default: return "unknown"; } } // Helper function to extract d tag value from tags array const char* extract_d_tag_value(cJSON* tags) { if (!tags || !cJSON_IsArray(tags)) { return NULL; } cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 2) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); cJSON* tag_value = cJSON_GetArrayItem(tag, 1); if (cJSON_IsString(tag_name) && cJSON_IsString(tag_value)) { const char* name = cJSON_GetStringValue(tag_name); if (name && strcmp(name, "d") == 0) { return cJSON_GetStringValue(tag_value); } } } } return NULL; } // Insert denormalized tags into event_tags table for fast indexed lookups int store_event_tags(const char* event_id, cJSON* tags) { return db_store_event_tags_cjson(event_id, tags); } // Core event storage path. // Returns: // 0 = inserted into DB // 1 = handled without insert (duplicate or ephemeral) // -1 = failure int store_event_core(cJSON* event) { if (!db_is_available() || !event) { return -1; } // Extract event fields cJSON* id = cJSON_GetObjectItemCaseSensitive(event, "id"); cJSON* pubkey = cJSON_GetObjectItemCaseSensitive(event, "pubkey"); cJSON* created_at = cJSON_GetObjectItemCaseSensitive(event, "created_at"); cJSON* kind = cJSON_GetObjectItemCaseSensitive(event, "kind"); cJSON* content = cJSON_GetObjectItemCaseSensitive(event, "content"); cJSON* sig = cJSON_GetObjectItemCaseSensitive(event, "sig"); cJSON* tags = cJSON_GetObjectItemCaseSensitive(event, "tags"); if (!id || !pubkey || !created_at || !kind || !content || !sig) { DEBUG_ERROR("Invalid event - missing required fields"); return -1; } // Classify event type event_type_t type = classify_event_kind((int)cJSON_GetNumberValue(kind)); // EPHEMERAL EVENTS (kinds 20000-29999) should NOT be stored if (type == EVENT_TYPE_EPHEMERAL) { DEBUG_LOG("Ephemeral event (kind %d) - broadcasting only, not storing", (int)cJSON_GetNumberValue(kind)); return 1; } // Serialize tags to JSON (use empty array if no tags) char* tags_json = NULL; if (tags && cJSON_IsArray(tags)) { tags_json = cJSON_Print(tags); } else { tags_json = strdup("[]"); } if (!tags_json) { DEBUG_ERROR("Failed to serialize tags to JSON"); return -1; } // Serialize full event JSON for fast retrieval (use PrintUnformatted for compact storage) char* event_json = cJSON_PrintUnformatted(event); if (!event_json) { DEBUG_ERROR("Failed to serialize event to JSON"); free(tags_json); return -1; } thread_pool_store_event_payload_t payload; memset(&payload, 0, sizeof(payload)); payload.id = (char*)cJSON_GetStringValue(id); payload.pubkey = (char*)cJSON_GetStringValue(pubkey); payload.created_at = (long long)cJSON_GetNumberValue(created_at); payload.kind = (int)cJSON_GetNumberValue(kind); payload.event_type = (char*)event_type_to_string(type); payload.content = (char*)cJSON_GetStringValue(content); payload.sig = (char*)cJSON_GetStringValue(sig); payload.tags_json = tags_json; payload.event_json = event_json; thread_pool_store_event_result_t tp_result; memset(&tp_result, 0, sizeof(tp_result)); if (thread_pool_execute_store_event_sync(&payload, &tp_result) != 0) { DEBUG_ERROR("Failed to execute event insert operation"); free(tags_json); free(event_json); return -1; } int rc = tp_result.step_rc; int extended_errcode = tp_result.extended_errcode; if (rc != DB_DONE) { const char* err_msg = db_last_error(); if (rc != DB_CONSTRAINT) { DEBUG_ERROR("INSERT failed: rc=%d, extended_errcode=%d, msg=%s", rc, extended_errcode, err_msg); } } if (rc != DB_DONE) { if (rc == DB_CONSTRAINT) { DEBUG_WARN("Event already exists in database"); // Add TRACE level debug to show both events if (g_debug_level >= DEBUG_LEVEL_TRACE) { // Get the existing event from database cJSON* existing_event = retrieve_event(cJSON_GetStringValue(id)); if (existing_event) { char* existing_json = cJSON_Print(existing_event); DEBUG_TRACE("EXISTING EVENT: %s", existing_json ? existing_json : "NULL"); free(existing_json); cJSON_Delete(existing_event); } else { DEBUG_TRACE("EXISTING EVENT: Could not retrieve existing event"); } // Show the event we're trying to insert char* new_json = cJSON_Print(event); DEBUG_TRACE("NEW EVENT: %s", new_json ? new_json : "NULL"); free(new_json); } free(tags_json); free(event_json); return 1; } char error_msg[256]; snprintf(error_msg, sizeof(error_msg), "Failed to insert event: %s", db_last_error()); DEBUG_ERROR(error_msg); free(tags_json); free(event_json); return -1; } free(tags_json); free(event_json); return 0; } // Main-thread-only post-store follow-up actions. void store_event_post_actions(cJSON* event) { if (!event) { return; } // Call monitoring hook after successful event storage monitoring_on_event_stored(); // Check if this is a kind 3 event from the admin — trigger WoT sync cJSON* kind_obj = cJSON_GetObjectItemCaseSensitive(event, "kind"); cJSON* pubkey_obj = cJSON_GetObjectItemCaseSensitive(event, "pubkey"); if (kind_obj && pubkey_obj && cJSON_GetNumberValue(kind_obj) == 3) { int wot_level = get_config_int("wot_enabled", 0); if (wot_level > 0) { const char* admin_pubkey = get_config_value("admin_pubkey"); if (admin_pubkey && strcmp(cJSON_GetStringValue(pubkey_obj), admin_pubkey) == 0) { DEBUG_INFO("Admin kind 3 event stored — triggering WoT sync"); extern int wot_sync_from_admin_kind3(void); wot_sync_from_admin_kind3(); } if (admin_pubkey) free((char*)admin_pubkey); } } } // Backward-compatible wrapper for synchronous call sites. int store_event(cJSON* event) { int core_rc = store_event_core(event); if (core_rc < 0) { return -1; } if (core_rc == 0) { store_event_post_actions(event); } return 0; } // Fast duplicate check: returns 1 if event ID already exists in DB, 0 if not. // Uses the primary key index — single B-tree lookup, ~10μs. // Call this BEFORE signature verification to skip expensive crypto on duplicates. int event_id_exists_in_db(const char* event_id) { if (!db_is_available() || !event_id || strlen(event_id) != 64) return 0; int exists = 0; if (db_event_id_exists(event_id, &exists) != 0) { return 0; } return exists; } // Populate event_tags from existing events (run once at startup) int populate_event_tags_from_existing(void) { return db_populate_event_tags_from_existing(); } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // EVENT STORAGE AND RETRIEVAL ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// cJSON* retrieve_event(const char* event_id) { return db_retrieve_event_by_id(event_id); } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // SUBSCRIPTION HANDLERS ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// /** * Check if filters contain only kind 99999 (NDK ping) * Returns 1 if all filters only request kind 99999, 0 otherwise */ static int is_only_kind_99999_request(cJSON* filters) { if (!filters || !cJSON_IsArray(filters)) { return 0; } int filter_count = cJSON_GetArraySize(filters); if (filter_count == 0) { return 0; } for (int i = 0; i < filter_count; i++) { cJSON* filter = cJSON_GetArrayItem(filters, i); if (!filter || !cJSON_IsObject(filter)) { return 0; } cJSON* kinds = cJSON_GetObjectItemCaseSensitive(filter, "kinds"); if (!kinds || !cJSON_IsArray(kinds)) { // Filter has no kinds or kinds is not an array - not a pure 99999 request return 0; } int kinds_count = cJSON_GetArraySize(kinds); if (kinds_count == 0) { return 0; } for (int j = 0; j < kinds_count; j++) { cJSON* kind_item = cJSON_GetArrayItem(kinds, j); if (!cJSON_IsNumber(kind_item)) { return 0; } int kind_val = (int)cJSON_GetNumberValue(kind_item); if (kind_val != 99999) { return 0; } } } return 1; // All filters only contain kind 99999 } int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { if (!cJSON_IsArray(filters)) { DEBUG_ERROR("REQ filters is not an array"); return 0; } // Check if this is a kind 99999 (NDK ping) request - these should never count toward rate limiting int is_ndk_ping = is_only_kind_99999_request(filters); // EARLY SUBSCRIPTION LIMIT CHECK - Check limits BEFORE any processing if (pss) { time_t current_time = time(NULL); // Check if client is currently rate limited due to excessive failed attempts if (pss->rate_limit_until > current_time) { // NDK ping (kind 99999) should not be blocked by rate limiting at all if (is_ndk_ping) { // Allow the request through - it will be handled normally and get a proper response // The subscription will be created but no events will match (which is correct) DEBUG_TRACE("Allowing kind 99999 NDK ping through despite rate limit"); // Fall through to normal processing (skip rate limit block) } else { char rate_limit_msg[256]; int remaining_seconds = (int)(pss->rate_limit_until - current_time); snprintf(rate_limit_msg, sizeof(rate_limit_msg), "Rate limited due to excessive failed subscription attempts. Try again in %d seconds.", remaining_seconds); // Send CLOSED notice for rate limiting cJSON* closed_msg = cJSON_CreateArray(); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("CLOSED")); cJSON_AddItemToArray(closed_msg, cJSON_CreateString(sub_id)); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("error: rate limited")); cJSON_AddItemToArray(closed_msg, cJSON_CreateString(rate_limit_msg)); char* closed_str = cJSON_Print(closed_msg); if (closed_str) { size_t closed_len = strlen(closed_str); unsigned char* buf = malloc(LWS_PRE + closed_len); if (buf) { memcpy(buf + LWS_PRE, closed_str, closed_len); lws_write(wsi, buf + LWS_PRE, closed_len, LWS_WRITE_TEXT); free(buf); } free(closed_str); } cJSON_Delete(closed_msg); // Do NOT increment rate limiting counters here - the client is already rate limited. // Incrementing counters while already blocked would extend the punishment indefinitely, // especially for benign requests like NDK's kind 99999 pings. return 0; } } // Check session subscription limits if (pss->subscription_count >= g_subscription_manager.max_subscriptions_per_client) { DEBUG_ERROR("Maximum subscriptions per client exceeded"); // Update rate limiting counters for failed attempt (but not for NDK ping) if (!is_ndk_ping) { pss->failed_subscription_attempts++; pss->last_failed_attempt = current_time; pss->consecutive_failures++; } // Implement progressive backoff: 1s, 5s, 30s, 300s (5min) based on consecutive failures int backoff_seconds = 1; if (pss->consecutive_failures >= 10) backoff_seconds = 300; // 5 minutes else if (pss->consecutive_failures >= 5) backoff_seconds = 30; // 30 seconds else if (pss->consecutive_failures >= 3) backoff_seconds = 5; // 5 seconds pss->rate_limit_until = current_time + backoff_seconds; // Send CLOSED notice with backoff information cJSON* closed_msg = cJSON_CreateArray(); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("CLOSED")); cJSON_AddItemToArray(closed_msg, cJSON_CreateString(sub_id)); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("error: too many subscriptions")); char backoff_msg[256]; snprintf(backoff_msg, sizeof(backoff_msg), "Maximum subscriptions per client exceeded. Backoff for %d seconds.", backoff_seconds); cJSON_AddItemToArray(closed_msg, cJSON_CreateString(backoff_msg)); char* closed_str = cJSON_Print(closed_msg); if (closed_str) { size_t closed_len = strlen(closed_str); unsigned char* buf = malloc(LWS_PRE + closed_len); if (buf) { memcpy(buf + LWS_PRE, closed_str, closed_len); lws_write(wsi, buf + LWS_PRE, closed_len, LWS_WRITE_TEXT); free(buf); } free(closed_str); } cJSON_Delete(closed_msg); return 0; } } // Parameter binding helpers char** bind_params = NULL; int bind_param_count = 0; int bind_param_capacity = 0; // Check for kind 33334 configuration event requests BEFORE creating subscription int config_events_sent = 0; int has_config_request = 0; // Check if any filter requests kind 33334 (configuration events) for (int i = 0; i < cJSON_GetArraySize(filters); i++) { cJSON* filter = cJSON_GetArrayItem(filters, i); if (filter && cJSON_IsObject(filter)) { if (req_filter_requests_config_events(filter)) { has_config_request = 1; // Generate synthetic config event for this subscription cJSON* filters_array = cJSON_CreateArray(); cJSON_AddItemToArray(filters_array, cJSON_Duplicate(filter, 1)); cJSON* event_msg = generate_synthetic_config_event_for_subscription(sub_id, filters_array); if (event_msg) { char* msg_str = cJSON_Print(event_msg); if (msg_str) { size_t msg_len = strlen(msg_str); // Use proper message queue system instead of direct lws_write if (queue_message(wsi, NULL, msg_str, msg_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue config EVENT message"); } else { config_events_sent++; } free(msg_str); } cJSON_Delete(event_msg); } cJSON_Delete(filters_array); break; // Only generate once per subscription } } } // If only config events were requested, we can return early after sending EOSE // But still create the subscription for future config updates // Create persistent subscription subscription_t* subscription = create_subscription(sub_id, wsi, filters, pss ? pss->client_ip : "unknown"); if (!subscription) { DEBUG_ERROR("Failed to create subscription"); return has_config_request ? config_events_sent : 0; } // Add to global manager if (add_subscription_to_manager(subscription) != 0) { DEBUG_ERROR("Failed to add subscription to global manager"); free_subscription(subscription); // Send CLOSED notice cJSON* closed_msg = cJSON_CreateArray(); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("CLOSED")); cJSON_AddItemToArray(closed_msg, cJSON_CreateString(sub_id)); cJSON_AddItemToArray(closed_msg, cJSON_CreateString("error: subscription limit reached")); char* closed_str = cJSON_Print(closed_msg); if (closed_str) { size_t closed_len = strlen(closed_str); // Use proper message queue system instead of direct lws_write if (queue_message(wsi, pss, closed_str, closed_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue CLOSED message"); } free(closed_str); } cJSON_Delete(closed_msg); // Update rate limiting counters for failed attempt (global limit reached) // Do not count NDK ping (kind 99999) toward rate limiting if (pss && !is_ndk_ping) { time_t current_time = time(NULL); pss->failed_subscription_attempts++; pss->last_failed_attempt = current_time; pss->consecutive_failures++; } return has_config_request ? config_events_sent : 0; } // Add to session's subscription list (if session data available) if (pss) { pthread_mutex_lock(&pss->session_lock); subscription->session_next = pss->subscriptions; pss->subscriptions = subscription; pss->subscription_count++; pthread_mutex_unlock(&pss->session_lock); } int events_sent = config_events_sent; // Start with synthetic config events int submitted_jobs = 0; int expiration_enabled = get_config_bool("expiration_enabled", 1); int filter_responses = get_config_bool("expiration_filter", 1); time_t query_now = time(NULL); req_async_state_t* async_state = req_async_state_create(sub_id, wsi); if (!async_state) { DEBUG_ERROR("Failed to allocate async REQ state"); free_bind_params(bind_params, bind_param_count); return events_sent; } // Process each filter in the array for (int i = 0; i < cJSON_GetArraySize(filters); i++) { cJSON* filter = cJSON_GetArrayItem(filters, i); if (!filter || !cJSON_IsObject(filter)) { DEBUG_WARN("Invalid filter object"); continue; } // NIP-01: limit:0 means return zero stored events — skip SQL entirely, just send EOSE cJSON* limit_check = cJSON_GetObjectItemCaseSensitive(filter, "limit"); if (limit_check && cJSON_IsNumber(limit_check) && (int)cJSON_GetNumberValue(limit_check) == 0) { DEBUG_LOG("Filter has limit:0 — skipping query, sending EOSE immediately (NIP-01 compliance)"); continue; } // Reset bind params for this filter free_bind_params(bind_params, bind_param_count); bind_params = NULL; bind_param_count = 0; bind_param_capacity = 0; // Build SQL query based on filter - exclude ephemeral events (kinds 20000-29999) from historical queries // Select event_json for fast retrieval (no JSON reconstruction needed) char sql[1408] = "SELECT event_json FROM events WHERE 1=1 AND (kind < 20000 OR kind >= 30000)"; char* sql_ptr = sql + strlen(sql); int remaining = sizeof(sql) - strlen(sql); // Phase 3: push expiration filtering into SQL using indexed event_tags table. if (expiration_enabled && filter_responses) { snprintf(sql_ptr, remaining, " AND NOT EXISTS (SELECT 1 FROM event_tags et_exp " "WHERE et_exp.event_id = events.id " "AND et_exp.tag_name = ? " "AND CAST(et_exp.tag_value AS INTEGER) <= ?)"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, "expiration"); char now_buf[32]; snprintf(now_buf, sizeof(now_buf), "%lld", (long long)query_now); add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, now_buf); } // Handle kinds filter cJSON* kinds = cJSON_GetObjectItemCaseSensitive(filter, "kinds"); if (kinds && cJSON_IsArray(kinds)) { int kind_count = cJSON_GetArraySize(kinds); if (kind_count > 0) { snprintf(sql_ptr, remaining, " AND kind IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int k = 0; k < kind_count; k++) { cJSON* kind = cJSON_GetArrayItem(kinds, k); if (cJSON_IsNumber(kind)) { if (k > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "%d", (int)cJSON_GetNumberValue(kind)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } // Handle authors filter cJSON* authors = cJSON_GetObjectItemCaseSensitive(filter, "authors"); if (authors && cJSON_IsArray(authors)) { int author_count = 0; // Count valid authors for (int a = 0; a < cJSON_GetArraySize(authors); a++) { cJSON* author = cJSON_GetArrayItem(authors, a); if (cJSON_IsString(author)) { author_count++; } } if (author_count > 0) { snprintf(sql_ptr, remaining, " AND pubkey IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int a = 0; a < author_count; a++) { if (a > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add author values to bind params for (int a = 0; a < cJSON_GetArraySize(authors); a++) { cJSON* author = cJSON_GetArrayItem(authors, a); if (cJSON_IsString(author)) { add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, cJSON_GetStringValue(author)); } } } } // Handle ids filter cJSON* ids = cJSON_GetObjectItemCaseSensitive(filter, "ids"); if (ids && cJSON_IsArray(ids)) { int id_count = 0; // Count valid ids for (int j = 0; j < cJSON_GetArraySize(ids); j++) { cJSON* id = cJSON_GetArrayItem(ids, j); if (cJSON_IsString(id)) { id_count++; } } if (id_count > 0) { snprintf(sql_ptr, remaining, " AND id IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int j = 0; j < id_count; j++) { if (j > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add id values to bind params for (int j = 0; j < cJSON_GetArraySize(ids); j++) { cJSON* id = cJSON_GetArrayItem(ids, j); if (cJSON_IsString(id)) { add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, cJSON_GetStringValue(id)); } } } } // Handle tag filters (#e, #p, #t, etc.) cJSON* filter_item = NULL; cJSON_ArrayForEach(filter_item, filter) { const char* filter_key = filter_item->string; if (filter_key && filter_key[0] == '#' && strlen(filter_key) > 1) { // This is a tag filter like "#e", "#p", etc. const char* tag_name = filter_key + 1; // Get the tag name (e, p, t, type, etc.) if (cJSON_IsArray(filter_item)) { int tag_value_count = 0; // Count valid tag values for (int j = 0; j < cJSON_GetArraySize(filter_item); j++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, j); if (cJSON_IsString(tag_value)) { tag_value_count++; } } if (tag_value_count > 0) { // Use indexed event_tags table instead of json_each() snprintf(sql_ptr, remaining, " AND id IN (SELECT event_id FROM event_tags WHERE tag_name = ? AND tag_value IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int j = 0; j < tag_value_count; j++) { if (j > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, "))"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add tag name and values to bind params add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, tag_name); for (int j = 0; j < cJSON_GetArraySize(filter_item); j++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, j); if (cJSON_IsString(tag_value)) { add_bind_param(&bind_params, &bind_param_count, &bind_param_capacity, cJSON_GetStringValue(tag_value)); } } } } } } // Handle search filter (NIP-50) cJSON* search = cJSON_GetObjectItemCaseSensitive(filter, "search"); if (search && cJSON_IsString(search)) { const char* search_term = cJSON_GetStringValue(search); if (search_term && strlen(search_term) > 0) { // Search in both content and tag values using LIKE // Escape single quotes in search term for SQL safety char escaped_search[256]; size_t escaped_len = 0; for (size_t j = 0; search_term[j] && escaped_len < sizeof(escaped_search) - 1; j++) { if (search_term[j] == '\'') { escaped_search[escaped_len++] = '\''; escaped_search[escaped_len++] = '\''; } else { escaped_search[escaped_len++] = search_term[j]; } } escaped_search[escaped_len] = '\0'; // Add search conditions for content and tags // Use tags LIKE to search within the JSON string representation of tags snprintf(sql_ptr, remaining, " AND (content LIKE '%%%s%%' OR tags LIKE '%%\"%s\"%%')", escaped_search, escaped_search); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } // Handle since filter cJSON* since = cJSON_GetObjectItemCaseSensitive(filter, "since"); if (since && cJSON_IsNumber(since)) { snprintf(sql_ptr, remaining, " AND created_at >= %ld", (long)cJSON_GetNumberValue(since)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } // Handle until filter cJSON* until = cJSON_GetObjectItemCaseSensitive(filter, "until"); if (until && cJSON_IsNumber(until)) { snprintf(sql_ptr, remaining, " AND created_at <= %ld", (long)cJSON_GetNumberValue(until)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } // Add ordering and limit snprintf(sql_ptr, remaining, " ORDER BY created_at DESC"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Handle limit filter cJSON* limit = cJSON_GetObjectItemCaseSensitive(filter, "limit"); if (limit && cJSON_IsNumber(limit)) { int limit_val = (int)cJSON_GetNumberValue(limit); if (limit_val > 0 && limit_val <= 5000) { snprintf(sql_ptr, remaining, " LIMIT %d", limit_val); } } else { // Default limit to prevent excessive queries snprintf(sql_ptr, remaining, " LIMIT 500"); } // Submit async REQ query (results processed on lws thread via completion queue) if (submit_req_query_async(async_state, sql, (const char**)bind_params, bind_param_count) != 0) { DEBUG_ERROR("Failed to submit async REQ query for subscription %s", sub_id); continue; } submitted_jobs++; if (pss) { pss->db_queries_executed++; } } // Cleanup bind params free_bind_params(bind_params, bind_param_count); if (submitted_jobs == 0) { req_async_state_free(async_state); return events_sent; } return HANDLE_REQ_ASYNC_PENDING; } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // ADMIN EVENT AUTHORIZATION ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Enhanced admin event authorization function int is_authorized_admin_event(cJSON* event, char* error_buffer, size_t error_buffer_size) { if (!event || !error_buffer) { if (error_buffer && error_buffer_size > 0) { snprintf(error_buffer, error_buffer_size, "Invalid parameters for admin authorization"); } return -1; } // Step 1: Verify event kind is admin type cJSON *kind_json = cJSON_GetObjectItemCaseSensitive(event, "kind"); if (!kind_json || !cJSON_IsNumber(kind_json)) { snprintf(error_buffer, error_buffer_size, "Missing or invalid event kind"); return -1; } int event_kind = kind_json->valueint; if (event_kind != 23456) { snprintf(error_buffer, error_buffer_size, "Event kind %d is not an admin event type", event_kind); return -1; } // Step 2: Check if event targets this relay (look for 'p' tag with our relay pubkey) cJSON *tags = cJSON_GetObjectItemCaseSensitive(event, "tags"); if (!tags || !cJSON_IsArray(tags)) { // No tags array - treat as regular event for different relay snprintf(error_buffer, error_buffer_size, "Admin event not targeting this relay (no tags)"); return -1; } int targets_this_relay = 0; const char* relay_pubkey = get_config_value("relay_pubkey"); cJSON *tag; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag)) { cJSON *tag_name = cJSON_GetArrayItem(tag, 0); cJSON *tag_value = cJSON_GetArrayItem(tag, 1); if (tag_name && cJSON_IsString(tag_name) && tag_value && cJSON_IsString(tag_value) && strcmp(tag_name->valuestring, "p") == 0) { // Compare with our relay pubkey if (relay_pubkey && strcmp(tag_value->valuestring, relay_pubkey) == 0) { targets_this_relay = 1; break; } } } } if (relay_pubkey) free((char*)relay_pubkey); if (!targets_this_relay) { // Admin event for different relay - not an error, just not for us snprintf(error_buffer, error_buffer_size, "Admin event not targeting this relay"); return -1; } // Step 3: Verify admin signature authorization cJSON *pubkey_json = cJSON_GetObjectItemCaseSensitive(event, "pubkey"); if (!pubkey_json || !cJSON_IsString(pubkey_json)) { DEBUG_WARN("Unauthorized admin event attempt: missing or invalid pubkey"); snprintf(error_buffer, error_buffer_size, "Unauthorized admin event attempt: missing pubkey"); return -1; } // Get admin pubkey from configuration const char* admin_pubkey = get_config_value("admin_pubkey"); if (!admin_pubkey || strlen(admin_pubkey) == 0) { DEBUG_WARN("Unauthorized admin event attempt: no admin pubkey configured"); snprintf(error_buffer, error_buffer_size, "Unauthorized admin event attempt: no admin configured"); if (admin_pubkey) free((char*)admin_pubkey); return -1; } // Compare pubkeys if (strcmp(pubkey_json->valuestring, admin_pubkey) != 0) { DEBUG_WARN("Unauthorized admin event attempt: pubkey mismatch"); char warning_msg[256]; snprintf(warning_msg, sizeof(warning_msg), "Unauthorized admin event attempt from pubkey: %.32s...", pubkey_json->valuestring); DEBUG_WARN(warning_msg); DEBUG_INFO("DEBUG: Pubkey comparison failed - event pubkey != admin pubkey"); snprintf(error_buffer, error_buffer_size, "Unauthorized admin event attempt: invalid admin pubkey"); free((char*)admin_pubkey); return -1; } // Step 4: Verify event signature if (nostr_verify_event_signature(event) != 0) { DEBUG_WARN("Unauthorized admin event attempt: invalid signature"); snprintf(error_buffer, error_buffer_size, "Unauthorized admin event attempt: signature verification failed"); free((char*)admin_pubkey); return -1; } free((char*)admin_pubkey); // All checks passed - authorized admin event return 0; } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // MAIN PROGRAM ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // Print usage information void print_usage(const char* program_name) { printf("Usage: %s [OPTIONS]\n", program_name); printf("\n"); printf("C Nostr Relay Server - Event-Based Configuration\n"); printf("\n"); printf("Options:\n"); printf(" -h, --help Show this help message\n"); printf(" -v, --version Show version information\n"); printf(" -p, --port PORT Override relay port (first-time startup and existing relay restarts)\n"); printf(" --strict-port Fail if exact port is unavailable (no port increment)\n"); printf(" -a, --admin-pubkey KEY Override admin public key (64-char hex or npub)\n"); printf(" -r, --relay-privkey KEY Override relay private key (64-char hex or nsec)\n"); printf(" --debug-level=N Set debug output level (0-5, default: 0)\n"); printf(" 0=none, 1=errors, 2=warnings, 3=info, 4=debug, 5=trace\n"); printf("\n"); printf("Configuration:\n"); printf(" This relay uses event-based configuration stored in the database.\n"); printf(" On first startup, keys are automatically generated and printed once.\n"); printf(" Command line options like --port apply during first-time setup and existing relay restarts.\n"); printf(" After initial setup, all configuration is managed via database events.\n"); printf(" Database file: .db (created automatically)\n"); printf("\n"); printf("Port Binding:\n"); printf(" Default: Try up to 10 consecutive ports if requested port is busy\n"); printf(" --strict-port: Fail immediately if exact requested port is unavailable\n"); printf(" --strict-port works with any custom port specified via -p or --port\n"); printf("\n"); printf("Examples:\n"); printf(" %s # Start relay (auto-configure on first run)\n", program_name); printf(" %s -p 8080 # First-time setup with port 8080\n", program_name); printf(" %s --port 9000 # First-time setup with port 9000\n", program_name); printf(" %s --strict-port # Fail if default port 8888 is unavailable\n", program_name); printf(" %s -p 8080 --strict-port # Fail if port 8080 is unavailable\n", program_name); printf(" %s --help # Show this help\n", program_name); printf(" %s --version # Show version info\n", program_name); printf("\n"); } // Print version information void print_version() { printf("C Nostr Relay Server %s\n", CRELAY_VERSION); printf("Event-based configuration system\n"); printf("Built with nostr_core_lib integration\n"); printf("\n"); } int main(int argc, char* argv[]) { // Initialize CLI options structure cli_options_t cli_options = { .port_override = -1, // -1 = not set .admin_pubkey_override = {0}, // Empty string = not set .relay_privkey_override = {0}, // Empty string = not set .strict_port = 0, // 0 = allow port increment (default) .debug_level = 0 // 0 = no debug output (default) }; // Parse command line arguments for (int i = 1; i < argc; i++) { if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) { print_usage(argv[0]); return 0; } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--version") == 0) { print_version(); return 0; } else if (strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) { // Port override option if (i + 1 >= argc) { DEBUG_ERROR("Port option requires a value. Use --help for usage information."); print_usage(argv[0]); return 1; } // Parse port number char* endptr; long port = strtol(argv[i + 1], &endptr, 10); if (endptr == argv[i + 1] || *endptr != '\0' || port < 1 || port > 65535) { DEBUG_ERROR("Invalid port number. Port must be between 1 and 65535."); print_usage(argv[0]); return 1; } cli_options.port_override = (int)port; i++; // Skip the port argument char port_msg[128]; snprintf(port_msg, sizeof(port_msg), "Port override specified: %d", cli_options.port_override); } else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--admin-pubkey") == 0) { // Admin public key override option if (i + 1 >= argc) { DEBUG_ERROR("Admin pubkey option requires a value. Use --help for usage information."); print_usage(argv[0]); return 1; } const char* input_key = argv[i + 1]; char decoded_key[65] = {0}; // Buffer for decoded hex key // Try to decode the input as either hex or npub format unsigned char pubkey_bytes[32]; if (nostr_decode_npub(input_key, pubkey_bytes) == NOSTR_SUCCESS) { // Convert bytes back to hex string char* hex_ptr = decoded_key; for (int j = 0; j < 32; j++) { sprintf(hex_ptr, "%02x", pubkey_bytes[j]); hex_ptr += 2; } } else { DEBUG_ERROR("Invalid admin public key format. Must be 64 hex characters or valid npub format."); print_usage(argv[0]); return 1; } strncpy(cli_options.admin_pubkey_override, decoded_key, sizeof(cli_options.admin_pubkey_override) - 1); cli_options.admin_pubkey_override[sizeof(cli_options.admin_pubkey_override) - 1] = '\0'; i++; // Skip the key argument } else if (strcmp(argv[i], "-r") == 0 || strcmp(argv[i], "--relay-privkey") == 0) { // Relay private key override option if (i + 1 >= argc) { DEBUG_ERROR("Relay privkey option requires a value. Use --help for usage information."); print_usage(argv[0]); return 1; } const char* input_key = argv[i + 1]; char decoded_key[65] = {0}; // Buffer for decoded hex key // Try to decode the input as either hex or nsec format unsigned char privkey_bytes[32]; if (nostr_decode_nsec(input_key, privkey_bytes) == NOSTR_SUCCESS) { // Convert bytes back to hex string char* hex_ptr = decoded_key; for (int j = 0; j < 32; j++) { sprintf(hex_ptr, "%02x", privkey_bytes[j]); hex_ptr += 2; } } else { DEBUG_ERROR("Invalid relay private key format. Must be 64 hex characters or valid nsec format."); print_usage(argv[0]); return 1; } strncpy(cli_options.relay_privkey_override, decoded_key, sizeof(cli_options.relay_privkey_override) - 1); cli_options.relay_privkey_override[sizeof(cli_options.relay_privkey_override) - 1] = '\0'; i++; // Skip the key argument } else if (strcmp(argv[i], "--strict-port") == 0) { // Strict port mode option cli_options.strict_port = 1; } else if (strncmp(argv[i], "--debug-level=", 14) == 0) { // Debug level option char* endptr; int debug_level = (int)strtol(argv[i] + 14, &endptr, 10); if (endptr == argv[i] + 14 || *endptr != '\0' || debug_level < 0 || debug_level > 5) { DEBUG_ERROR("Invalid debug level. Debug level must be between 0 and 5."); print_usage(argv[0]); return 1; } cli_options.debug_level = debug_level; } else { DEBUG_ERROR("Unknown argument. Use --help for usage information."); print_usage(argv[0]); return 1; } } // Initialize debug system debug_init(cli_options.debug_level); // Set up signal handlers signal(SIGINT, signal_handler); signal(SIGTERM, signal_handler); // Print version at startup (always, regardless of debug level) fprintf(stderr, "[RELAY_VERSION] C Nostr Relay Server %s\n", CRELAY_VERSION); fflush(stderr); printf(BLUE BOLD "=== C Nostr Relay Server ===" RESET "\n"); DEBUG_TRACE("Starting main initialization sequence"); // Initialize nostr library FIRST (required for key generation and event creation) if (nostr_init() != 0) { DEBUG_ERROR("Failed to initialize nostr library"); return 1; } DEBUG_LOG("Nostr library initialized"); // Route nostr_core_lib logs into relay logging pipeline (stdout/stderr -> journald or relay.log) nostr_set_log_callback(nostr_log_cb, NULL); nostr_log_level_t nostr_level = nostr_log_level_from_debug_level(cli_options.debug_level); nostr_set_log_level(nostr_level); DEBUG_INFO("Nostr callback logging initialized at level %d", (int)nostr_level); // Check if this is first-time startup or existing relay if (is_first_time_startup()) { DEBUG_LOG("First-time startup detected"); // Initialize event-based configuration system if (init_configuration_system(NULL, NULL) != 0) { DEBUG_ERROR("Failed to initialize event-based configuration system"); nostr_cleanup(); return 1; } // Run first-time startup sequence (generates keys, sets up database path, but doesn't store private key yet) char admin_pubkey[65] = {0}; char relay_pubkey[65] = {0}; char relay_privkey[65] = {0}; if (first_time_startup_sequence(&cli_options, admin_pubkey, relay_pubkey, relay_privkey) != 0) { DEBUG_ERROR("Failed to complete first-time startup sequence"); cleanup_configuration_system(); nostr_cleanup(); return 1; } // Initialize database with the generated relay pubkey DEBUG_TRACE("Initializing database for first-time startup"); if (init_database(g_database_path) != 0) { DEBUG_ERROR("Failed to initialize database after first-time setup"); cleanup_configuration_system(); nostr_cleanup(); return 1; } DEBUG_LOG("Database initialized for first-time startup"); // DEBUG_GUARD_START if (g_debug_level >= DEBUG_LEVEL_DEBUG) { int row_count = 0; if (db_get_config_row_count(&row_count) == 0) { DEBUG_LOG("Config table row count after init_database() (first-time): %d", row_count); } } // DEBUG_GUARD_END // Now that database is available, populate the complete config table atomically // BUG FIX: Use the pubkeys returned from first_time_startup_sequence instead of trying to read from empty database DEBUG_LOG("Using pubkeys from first-time startup sequence for config population"); DEBUG_LOG("admin_pubkey from startup: %s", admin_pubkey); DEBUG_LOG("relay_pubkey from startup: %s", relay_pubkey); if (populate_all_config_values_atomic(admin_pubkey, relay_pubkey) != 0) { DEBUG_ERROR("Failed to populate complete config table"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } // Apply CLI overrides atomically (after complete config table exists) if (apply_cli_overrides_atomic(&cli_options) != 0) { DEBUG_ERROR("Failed to apply CLI overrides"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } // Now that database is available, store the relay private key securely if (relay_privkey[0] != '\0') { if (store_relay_private_key(relay_privkey) != 0) { DEBUG_ERROR("Failed to store relay private key securely after database initialization"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } } else { DEBUG_ERROR("Relay private key not available from first-time startup"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } } else { // Find existing database file char** existing_files = find_existing_db_files(); if (!existing_files || !existing_files[0]) { DEBUG_ERROR("No existing relay database found"); nostr_cleanup(); return 1; } // Extract relay pubkey from filename char* relay_pubkey = extract_pubkey_from_filename(existing_files[0]); if (!relay_pubkey) { DEBUG_ERROR("Failed to extract relay pubkey from database filename"); // Free the files array for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); return 1; } // Initialize event-based configuration system if (init_configuration_system(NULL, NULL) != 0) { DEBUG_ERROR("Failed to initialize event-based configuration system"); free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); return 1; } // Setup existing relay FIRST (sets database path) if (startup_existing_relay(relay_pubkey, &cli_options) != 0) { DEBUG_ERROR("Failed to setup existing relay"); cleanup_configuration_system(); free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); return 1; } // Initialize database with the database path set by startup_existing_relay() DEBUG_TRACE("Initializing existing database"); if (init_database(g_database_path) != 0) { DEBUG_ERROR("Failed to initialize existing database"); cleanup_configuration_system(); free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); return 1; } DEBUG_LOG("Existing database initialized"); // Keep relay_version in sync with the compiled binary version on every startup if (update_config_in_table("relay_version", CRELAY_VERSION) != 0) { DEBUG_ERROR("Failed to synchronize relay_version with compiled CRELAY_VERSION"); cleanup_configuration_system(); free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); close_database(); return 1; } DEBUG_INFO("Synchronized relay_version to %s", CRELAY_VERSION); // Apply CLI overrides atomically (now that database is initialized) if (apply_cli_overrides_atomic(&cli_options) != 0) { DEBUG_ERROR("Failed to apply CLI overrides for existing relay"); cleanup_configuration_system(); free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); nostr_cleanup(); close_database(); return 1; } // DEBUG_GUARD_START if (g_debug_level >= DEBUG_LEVEL_DEBUG) { int row_count = 0; if (db_get_config_row_count(&row_count) == 0) { DEBUG_LOG("Config table row count after init_database(): %d", row_count); } } // Free memory free(relay_pubkey); for (int i = 0; existing_files[i]; i++) { free(existing_files[i]); } free(existing_files); } // Verify database is now available if (!g_db) { DEBUG_ERROR("Database not available after initialization"); cleanup_configuration_system(); nostr_cleanup(); return 1; } // Pre-warm unified config cache while DB is still available. // This is critical because runtime strict mode sets g_db = NULL, // and admin authorization depends on relay_pubkey/admin_pubkey reads. if (reload_config_from_table() != 0) { DEBUG_ERROR("Failed to pre-warm configuration cache from config table"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } const char* prewarmed_relay_pubkey = get_config_value("relay_pubkey"); const char* prewarmed_admin_pubkey = get_config_value("admin_pubkey"); if (!prewarmed_relay_pubkey || strlen(prewarmed_relay_pubkey) != 64 || !prewarmed_admin_pubkey || strlen(prewarmed_admin_pubkey) != 64) { if (prewarmed_relay_pubkey) free((char*)prewarmed_relay_pubkey); if (prewarmed_admin_pubkey) free((char*)prewarmed_admin_pubkey); DEBUG_ERROR("Critical config pre-warm validation failed (relay_pubkey/admin_pubkey missing)"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } free((char*)prewarmed_relay_pubkey); free((char*)prewarmed_admin_pubkey); // Preload relay private key runtime cache while DB is still available. // Required for admin command decrypt/encrypt after strict mode sets g_db = NULL. if (preload_relay_private_key_cache() != 0) { DEBUG_ERROR("Failed to pre-warm relay private key cache"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } DEBUG_INFO("Configuration cache pre-warmed with relay_pubkey/admin_pubkey and relay private key"); // Configuration system is now fully initialized with event-based approach // All configuration is loaded from database events // Initialize unified request validator system if (ginxsom_request_validator_init(g_database_path, "c-relay") != 0) { DEBUG_ERROR("Failed to initialize unified request validator"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } // Initialize NIP-11 relay information init_relay_info(); // Initialize NIP-13 PoW configuration init_pow_config(); // Initialize NIP-40 expiration configuration init_expiration_config(); // Update subscription manager configuration update_subscription_manager_config(); // Initialize subscription manager mutexes if (pthread_mutex_init(&g_subscription_manager.subscriptions_lock, NULL) != 0) { DEBUG_ERROR("Failed to initialize subscription manager subscriptions lock"); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } if (pthread_mutex_init(&g_subscription_manager.ip_tracking_lock, NULL) != 0) { DEBUG_ERROR("Failed to initialize subscription manager IP tracking lock"); pthread_mutex_destroy(&g_subscription_manager.subscriptions_lock); cleanup_configuration_system(); nostr_cleanup(); close_database(); return 1; } // Initialize kind-based index for fast subscription lookup init_kind_index(); // Populate event_tags from existing events (for tag-based lookups) populate_event_tags_from_existing(); // Sync Web of Trust whitelist if enabled int wot_level = get_config_int("wot_enabled", 0); if (wot_level > 0) { DEBUG_INFO("WoT enabled (level %d) - syncing from admin's kind 3 event", wot_level); extern int wot_sync_from_admin_kind3(void); wot_sync_from_admin_kind3(); } // Cleanup orphaned subscriptions from previous runs cleanup_all_subscriptions_on_startup(); // Initialize IP ban table and load persisted state from database ip_ban_init(); 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", 1); 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", 1); 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"); } } // Phase 5 strict mode: remove main-thread fallback to global DB handle. // Runtime DB access must go through thread-bound worker connections. g_db = NULL; // 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(); cleanup_configuration_system(); // Cleanup subscription manager mutexes pthread_mutex_destroy(&g_subscription_manager.subscriptions_lock); pthread_mutex_destroy(&g_subscription_manager.ip_tracking_lock); nostr_set_log_callback(NULL, NULL); nostr_cleanup(); close_database(); if (result == 0) { } else { DEBUG_ERROR("Server shutdown with errors"); } return result; }