v2.1.1 - Wire REQ/COUNT/EVENT paths through thread pool sync helpers and stabilize filter limit tests

relay.pid

@@ -1 +1 @@
-504823
+513827

src/main.c

@@ -677,10 +677,6 @@ int store_event_tags(const char* event_id, cJSON* tags) {
 
 // Store event in database
 int store_event(cJSON* event) {
-    if (thread_pool_is_running()) {
-        DEBUG_TRACE("Thread pool scaffold active: store_event() still using synchronous DB path");
-    }
-
     if (!g_db || !event) {
         return -1;
     }
@@ -730,25 +726,30 @@ int store_event(cJSON* event) {
         return -1;
     }
     
-    int rc = DB_ERROR;
+    thread_pool_store_event_payload_t payload;
-    int extended_errcode = 0;
+    memset(&payload, 0, sizeof(payload));
-    if (db_insert_event_with_json(cJSON_GetStringValue(id),
+    payload.id = (char*)cJSON_GetStringValue(id);
-                                  cJSON_GetStringValue(pubkey),
+    payload.pubkey = (char*)cJSON_GetStringValue(pubkey);
-                                  (long long)cJSON_GetNumberValue(created_at),
+    payload.created_at = (long long)cJSON_GetNumberValue(created_at);
-                                  (int)cJSON_GetNumberValue(kind),
+    payload.kind = (int)cJSON_GetNumberValue(kind);
-                                  event_type_to_string(type),
+    payload.event_type = (char*)event_type_to_string(type);
-                                  cJSON_GetStringValue(content),
+    payload.content = (char*)cJSON_GetStringValue(content);
-                                  cJSON_GetStringValue(sig),
+    payload.sig = (char*)cJSON_GetStringValue(sig);
-                                  tags_json,
+    payload.tags_json = tags_json;
-                                  event_json,
+    payload.event_json = event_json;
-                                  &rc,
+
-                                  &extended_errcode) != 0) {
+    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) {
@@ -901,10 +902,6 @@ static int is_only_kind_99999_request(cJSON* filters) {
 }
 
 int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) {
-    if (thread_pool_is_running()) {
-        DEBUG_TRACE("Thread pool scaffold active: handle_req_message() still using synchronous DB path");
-    }
-
     if (!cJSON_IsArray(filters)) {
         DEBUG_ERROR("REQ filters is not an array");
         return 0;
@@ -1358,14 +1355,13 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
         struct timespec query_start, query_end;
         clock_gettime(CLOCK_MONOTONIC, &query_start);
         
-        // Execute query and send events
+        // Execute query through db thread pool helper and send events
-        db_stmt_t* stmt;
+        thread_pool_req_result_t* req_result = NULL;
-        int rc = db_prepare(sql, &stmt);
+        if (thread_pool_execute_req_sync(sql, (const char**)bind_params, bind_param_count, 5000, &req_result) != 0 || !req_result) {
-        if (rc != DB_OK) {
             char error_msg[256];
-            snprintf(error_msg, sizeof(error_msg), "Failed to prepare subscription query: %s", db_last_error());
+            snprintf(error_msg, sizeof(error_msg), "Failed to execute subscription query: %s", db_last_error());
             DEBUG_ERROR(error_msg);
-            
+
             // Log the failed query so we can see what SQL was generated
             if (g_debug_level >= DEBUG_LEVEL_DEBUG) {
                 time_t now = time(NULL);
@@ -1380,7 +1376,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
                         sql);
                 fflush(stderr);
             }
-            
+
             continue;
         }
 
@@ -1389,38 +1385,32 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
             pss->db_queries_executed++;
         }
 
-        // Bind parameters
-        for (int i = 0; i < bind_param_count; i++) {
-            db_bind_text_param(stmt, i + 1, bind_params[i]);
-        }
-
         // Cache config values outside the row loop (performance fix)
         int expiration_enabled = get_config_bool("expiration_enabled", 1);
         int filter_responses = get_config_bool("expiration_filter", 1);
 
         int row_count = 0;
-        while (db_step_stmt(stmt) == DB_ROW) {
+        for (int r = 0; r < req_result->row_count; r++) {
+            const char* event_json_str = req_result->event_json_rows[r];
             row_count++;
-            
+
             // Track rows returned for abuse detection
             if (pss) {
                 pss->db_rows_returned++;
             }
-            
+
-            // Get pre-serialized event JSON (no reconstruction needed!)
-            const char* event_json_str = db_column_text_value(stmt, 0);
             if (!event_json_str) {
                 DEBUG_ERROR("Event has NULL event_json field");
                 continue;
             }
-            
+
             // Parse event JSON only for expiration check
             cJSON* event = cJSON_Parse(event_json_str);
             if (!event) {
                 DEBUG_ERROR("Failed to parse event_json from database");
                 continue;
             }
-            
+
             // Check expiration filtering (NIP-40) at application level
             // (expiration_enabled and filter_responses are cached outside the loop)
             if (expiration_enabled && filter_responses) {
@@ -1431,7 +1421,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
                     continue;
                 }
             }
-            
+
             // Build EVENT message using zero-copy path: allocate with LWS_PRE prefix,
             // write directly, transfer ownership to queue — no memcpy.
             // Format: ["EVENT","<sub_id>",<event_json>]
@@ -1456,7 +1446,7 @@ int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, stru
             events_sent++;
         }
 
-        db_finalize_stmt(stmt);
+        thread_pool_free_req_result(req_result);
         
         // Stop query timing and log
         clock_gettime(CLOCK_MONOTONIC, &query_end);

src/main.h

@@ -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 1
-#define CRELAY_VERSION_PATCH 0
+#define CRELAY_VERSION_PATCH 1
-#define CRELAY_VERSION "v2.1.0"
+#define CRELAY_VERSION "v2.1.1"
 
 // Relay metadata (authoritative source for NIP-11 information)
 #define RELAY_NAME "C-Relay"

src/thread_pool.c

@@ -2,6 +2,7 @@
 
 #include "thread_pool.h"
 #include "debug.h"
+#include "db_ops.h"
 
 #include <pthread.h>
 #include <stdlib.h>
@@ -41,6 +42,86 @@ typedef struct {
 
 static thread_pool_state_t g_pool = {0};
 
+typedef struct {
+    pthread_mutex_t mutex;
+    pthread_cond_t cond;
+    int done;
+    thread_pool_status_t status;
+    void* result_data;
+    size_t result_size;
+} thread_pool_wait_ctx_t;
+
+static void wait_ctx_init(thread_pool_wait_ctx_t* ctx) {
+    memset(ctx, 0, sizeof(*ctx));
+    pthread_mutex_init(&ctx->mutex, NULL);
+    pthread_cond_init(&ctx->cond, NULL);
+}
+
+static void wait_ctx_destroy(thread_pool_wait_ctx_t* ctx) {
+    pthread_mutex_destroy(&ctx->mutex);
+    pthread_cond_destroy(&ctx->cond);
+}
+
+static void wait_ctx_result_cb(const thread_pool_result_t* result, void* user_ctx) {
+    thread_pool_wait_ctx_t* ctx = (thread_pool_wait_ctx_t*)user_ctx;
+    if (!ctx || !result) return;
+
+    pthread_mutex_lock(&ctx->mutex);
+    ctx->status = result->status;
+    ctx->result_data = result->result_data;
+    ctx->result_size = result->result_size;
+    ctx->done = 1;
+    pthread_cond_signal(&ctx->cond);
+    pthread_mutex_unlock(&ctx->mutex);
+}
+
+static void wait_ctx_wait(thread_pool_wait_ctx_t* ctx) {
+    pthread_mutex_lock(&ctx->mutex);
+    while (!ctx->done) {
+        pthread_cond_wait(&ctx->cond, &ctx->mutex);
+    }
+    pthread_mutex_unlock(&ctx->mutex);
+}
+
+static void free_count_payload(void* p) {
+    thread_pool_count_payload_t* payload = (thread_pool_count_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 void free_req_payload(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 void free_store_event_payload(void* p) {
+    thread_pool_store_event_payload_t* payload = (thread_pool_store_event_payload_t*)p;
+    if (!payload) return;
+    free(payload->id);
+    free(payload->pubkey);
+    free(payload->event_type);
+    free(payload->content);
+    free(payload->sig);
+    free(payload->tags_json);
+    free(payload->event_json);
+    free(payload);
+}
+
 static void queue_init(thread_pool_queue_t* q, int max_size) {
     memset(q, 0, sizeof(*q));
     q->max_size = (max_size > 0) ? max_size : 4096;
@@ -112,13 +193,19 @@ static void wake_event_loop(void) {
     }
 }
 
-static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_status_t status, const char* message) {
+static void complete_job_with_result(thread_pool_job_node_t* node,
+                                     thread_pool_status_t status,
+                                     const char* message,
+                                     void* result_data,
+                                     size_t result_size) {
     thread_pool_result_t result;
     memset(&result, 0, sizeof(result));
     result.job_id = node->job_id;
     result.type = node->job.type;
     result.status = status;
     result.session = node->job.session;
+    result.result_data = result_data;
+    result.result_size = result_size;
     result.message = message;
 
     if (node->job.result_cb) {
@@ -132,15 +219,167 @@ static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_s
     free(node);
 }
 
+static void execute_count_job(thread_pool_job_node_t* node) {
+    thread_pool_count_payload_t* payload = (thread_pool_count_payload_t*)node->job.payload;
+    if (!payload || !payload->sql) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "COUNT payload missing", NULL, 0);
+        return;
+    }
+
+    int count = 0;
+    if (db_count_with_sql(payload->sql, (const char**)payload->bind_params,
+                          payload->bind_param_count, &count) != 0) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "COUNT query failed", NULL, 0);
+        return;
+    }
+
+    thread_pool_count_result_t* out = calloc(1, sizeof(*out));
+    if (!out) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "COUNT result allocation failed", NULL, 0);
+        return;
+    }
+    out->count = count;
+
+    complete_job_with_result(node, THREAD_POOL_STATUS_OK,
+                             "COUNT query completed", out, sizeof(*out));
+}
+
+static void execute_req_job(thread_pool_job_node_t* node) {
+    thread_pool_req_payload_t* payload = (thread_pool_req_payload_t*)node->job.payload;
+    if (!payload || !payload->sql) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "REQ payload missing", NULL, 0);
+        return;
+    }
+
+    db_stmt_t* stmt = NULL;
+    if (db_prepare(payload->sql, &stmt) != DB_OK) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "REQ prepare failed", NULL, 0);
+        return;
+    }
+
+    for (int i = 0; i < payload->bind_param_count; i++) {
+        const char* v = (payload->bind_params && payload->bind_params[i]) ? payload->bind_params[i] : "";
+        (void)db_bind_text_param(stmt, i + 1, v);
+    }
+
+    int cap = (payload->row_limit > 0) ? payload->row_limit : 500;
+    if (cap < 1) cap = 1;
+
+    thread_pool_req_result_t* out = calloc(1, sizeof(*out));
+    if (!out) {
+        db_finalize_stmt(stmt);
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "REQ result allocation failed", NULL, 0);
+        return;
+    }
+
+    out->event_json_rows = calloc((size_t)cap, sizeof(char*));
+    if (!out->event_json_rows) {
+        db_finalize_stmt(stmt);
+        free(out);
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "REQ row array allocation failed", NULL, 0);
+        return;
+    }
+
+    int rc = DB_OK;
+    while ((rc = db_step_stmt(stmt)) == DB_ROW) {
+        if (out->row_count >= cap) break;
+        const char* row = db_column_text_value(stmt, 0);
+        if (!row) continue;
+
+        out->event_json_rows[out->row_count] = strdup(row);
+        if (!out->event_json_rows[out->row_count]) {
+            break;
+        }
+        out->row_count++;
+    }
+
+    db_finalize_stmt(stmt);
+
+    complete_job_with_result(node, THREAD_POOL_STATUS_OK,
+                             "REQ query completed", out, sizeof(*out));
+}
+
+static void execute_store_event_job(thread_pool_job_node_t* node) {
+    thread_pool_store_event_payload_t* payload = (thread_pool_store_event_payload_t*)node->job.payload;
+    if (!payload || !payload->id || !payload->pubkey || !payload->content || !payload->sig) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "STORE_EVENT payload missing required fields", NULL, 0);
+        return;
+    }
+
+    thread_pool_store_event_result_t* out = calloc(1, sizeof(*out));
+    if (!out) {
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "STORE_EVENT result allocation failed", NULL, 0);
+        return;
+    }
+
+    int step_rc = DB_ERROR;
+    int extended_errcode = 0;
+    if (db_insert_event_with_json(payload->id,
+                                  payload->pubkey,
+                                  payload->created_at,
+                                  payload->kind,
+                                  payload->event_type ? payload->event_type : "regular",
+                                  payload->content,
+                                  payload->sig,
+                                  payload->tags_json ? payload->tags_json : "[]",
+                                  payload->event_json ? payload->event_json : "{}",
+                                  &step_rc,
+                                  &extended_errcode) != 0) {
+        free(out);
+        complete_job_with_result(node, THREAD_POOL_STATUS_INTERNAL_ERROR,
+                                 "STORE_EVENT execution failed", NULL, 0);
+        return;
+    }
+
+    out->step_rc = step_rc;
+    out->extended_errcode = extended_errcode;
+    complete_job_with_result(node, THREAD_POOL_STATUS_OK,
+                             "STORE_EVENT completed", out, sizeof(*out));
+}
+
+static void execute_read_job(thread_pool_job_node_t* node) {
+    switch (node->job.type) {
+        case THREAD_POOL_JOB_REQ_QUERY:
+            execute_req_job(node);
+            break;
+        case THREAD_POOL_JOB_COUNT_QUERY:
+            execute_count_job(node);
+            break;
+        default:
+            complete_job_with_result(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
+                                     "Read job type not implemented", NULL, 0);
+            break;
+    }
+}
+
+static void execute_write_job(thread_pool_job_node_t* node) {
+    switch (node->job.type) {
+        case THREAD_POOL_JOB_STORE_EVENT:
+            execute_store_event_job(node);
+            break;
+        default:
+            complete_job_with_result(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
+                                     "Write job type not implemented", NULL, 0);
+            break;
+    }
+}
+
 static void* reader_worker_main(void* arg) {
     (void)arg;
     while (g_pool.running) {
         thread_pool_job_node_t* node = queue_pop(&g_pool.read_q);
         if (!node) break;
 
-        // Scaffold only: execution wiring is intentionally deferred.
+        execute_read_job(node);
-        complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
-                                 "Read worker scaffold active; execution not wired yet");
     }
     return NULL;
 }
@@ -151,9 +390,7 @@ static void* writer_worker_main(void* arg) {
         thread_pool_job_node_t* node = queue_pop(&g_pool.write_q);
         if (!node) break;
 
-        // Scaffold only: execution wiring is intentionally deferred.
+        execute_write_job(node);
-        complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
-                                 "Write worker scaffold active; execution not wired yet");
     }
     return NULL;
 }
@@ -270,6 +507,271 @@ thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint
     return submit_to_queue(&g_pool.write_q, job, out_job_id);
 }
 
+int thread_pool_execute_count_sync(const char* sql, const char** bind_params, int bind_param_count, int* out_count) {
+    if (!sql || !out_count) {
+        return -1;
+    }
+
+    if (!thread_pool_is_running()) {
+        return db_count_with_sql(sql, bind_params, bind_param_count, out_count);
+    }
+
+    thread_pool_count_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_count_payload(payload);
+        return -1;
+    }
+
+    if (bind_param_count > 0) {
+        payload->bind_params = calloc((size_t)bind_param_count, sizeof(char*));
+        if (!payload->bind_params) {
+            free_count_payload(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_count_payload(payload);
+                return -1;
+            }
+        }
+    }
+
+    thread_pool_wait_ctx_t wait_ctx;
+    wait_ctx_init(&wait_ctx);
+
+    thread_pool_job_t job;
+    memset(&job, 0, sizeof(job));
+    job.type = THREAD_POOL_JOB_COUNT_QUERY;
+    job.payload = payload;
+    job.payload_free = free_count_payload;
+    job.result_cb = wait_ctx_result_cb;
+    job.result_cb_ctx = &wait_ctx;
+
+    thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL);
+    if (submit_rc != THREAD_POOL_STATUS_OK) {
+        free_count_payload(payload);
+        wait_ctx_destroy(&wait_ctx);
+        return -1;
+    }
+
+    wait_ctx_wait(&wait_ctx);
+
+    int rc = -1;
+    if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) {
+        thread_pool_count_result_t* result = (thread_pool_count_result_t*)wait_ctx.result_data;
+        *out_count = result->count;
+        free(result);
+        rc = 0;
+    }
+
+    wait_ctx_destroy(&wait_ctx);
+    return rc;
+}
+
+int thread_pool_execute_req_sync(const char* sql, const char** bind_params, int bind_param_count,
+                                 int row_limit, thread_pool_req_result_t** out_result) {
+    if (!sql || !out_result) {
+        return -1;
+    }
+
+    if (!thread_pool_is_running()) {
+        db_stmt_t* stmt = NULL;
+        if (db_prepare(sql, &stmt) != DB_OK) {
+            return -1;
+        }
+
+        for (int i = 0; i < bind_param_count; i++) {
+            const char* v = (bind_params && bind_params[i]) ? bind_params[i] : "";
+            (void)db_bind_text_param(stmt, i + 1, v);
+        }
+
+        int cap = (row_limit > 0) ? row_limit : 500;
+        if (cap < 1) cap = 1;
+
+        thread_pool_req_result_t* result = calloc(1, sizeof(*result));
+        if (!result) {
+            db_finalize_stmt(stmt);
+            return -1;
+        }
+
+        result->event_json_rows = calloc((size_t)cap, sizeof(char*));
+        if (!result->event_json_rows) {
+            free(result);
+            db_finalize_stmt(stmt);
+            return -1;
+        }
+
+        int step_rc = DB_OK;
+        while ((step_rc = db_step_stmt(stmt)) == DB_ROW) {
+            if (result->row_count >= cap) break;
+            const char* row = db_column_text_value(stmt, 0);
+            if (!row) continue;
+
+            result->event_json_rows[result->row_count] = strdup(row);
+            if (!result->event_json_rows[result->row_count]) {
+                break;
+            }
+            result->row_count++;
+        }
+
+        db_finalize_stmt(stmt);
+        *out_result = result;
+        return 0;
+    }
+
+    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;
+    payload->row_limit = row_limit;
+    if (!payload->sql) {
+        free_req_payload(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(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(payload);
+                return -1;
+            }
+        }
+    }
+
+    thread_pool_wait_ctx_t wait_ctx;
+    wait_ctx_init(&wait_ctx);
+
+    thread_pool_job_t job;
+    memset(&job, 0, sizeof(job));
+    job.type = THREAD_POOL_JOB_REQ_QUERY;
+    job.payload = payload;
+    job.payload_free = free_req_payload;
+    job.result_cb = wait_ctx_result_cb;
+    job.result_cb_ctx = &wait_ctx;
+
+    thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL);
+    if (submit_rc != THREAD_POOL_STATUS_OK) {
+        free_req_payload(payload);
+        wait_ctx_destroy(&wait_ctx);
+        return -1;
+    }
+
+    wait_ctx_wait(&wait_ctx);
+
+    int rc = -1;
+    if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) {
+        *out_result = (thread_pool_req_result_t*)wait_ctx.result_data;
+        rc = 0;
+    }
+
+    wait_ctx_destroy(&wait_ctx);
+    return rc;
+}
+
+void thread_pool_free_req_result(thread_pool_req_result_t* result) {
+    if (!result) return;
+    if (result->event_json_rows) {
+        for (int i = 0; i < result->row_count; i++) {
+            free(result->event_json_rows[i]);
+        }
+        free(result->event_json_rows);
+    }
+    free(result);
+}
+
+int thread_pool_execute_store_event_sync(const thread_pool_store_event_payload_t* payload,
+                                         thread_pool_store_event_result_t* out_result) {
+    if (!payload || !out_result || !payload->id || !payload->pubkey || !payload->content || !payload->sig) {
+        return -1;
+    }
+
+    if (!thread_pool_is_running()) {
+        int step_rc = DB_ERROR;
+        int extended_errcode = 0;
+        if (db_insert_event_with_json(payload->id,
+                                      payload->pubkey,
+                                      payload->created_at,
+                                      payload->kind,
+                                      payload->event_type ? payload->event_type : "regular",
+                                      payload->content,
+                                      payload->sig,
+                                      payload->tags_json ? payload->tags_json : "[]",
+                                      payload->event_json ? payload->event_json : "{}",
+                                      &step_rc,
+                                      &extended_errcode) != 0) {
+            return -1;
+        }
+        out_result->step_rc = step_rc;
+        out_result->extended_errcode = extended_errcode;
+        return 0;
+    }
+
+    thread_pool_store_event_payload_t* payload_copy = calloc(1, sizeof(*payload_copy));
+    if (!payload_copy) return -1;
+
+    payload_copy->id = payload->id ? strdup(payload->id) : NULL;
+    payload_copy->pubkey = payload->pubkey ? strdup(payload->pubkey) : NULL;
+    payload_copy->created_at = payload->created_at;
+    payload_copy->kind = payload->kind;
+    payload_copy->event_type = payload->event_type ? strdup(payload->event_type) : strdup("regular");
+    payload_copy->content = payload->content ? strdup(payload->content) : NULL;
+    payload_copy->sig = payload->sig ? strdup(payload->sig) : NULL;
+    payload_copy->tags_json = payload->tags_json ? strdup(payload->tags_json) : strdup("[]");
+    payload_copy->event_json = payload->event_json ? strdup(payload->event_json) : strdup("{}");
+
+    if (!payload_copy->id || !payload_copy->pubkey || !payload_copy->event_type ||
+        !payload_copy->content || !payload_copy->sig || !payload_copy->tags_json || !payload_copy->event_json) {
+        free_store_event_payload(payload_copy);
+        return -1;
+    }
+
+    thread_pool_wait_ctx_t wait_ctx;
+    wait_ctx_init(&wait_ctx);
+
+    thread_pool_job_t job;
+    memset(&job, 0, sizeof(job));
+    job.type = THREAD_POOL_JOB_STORE_EVENT;
+    job.payload = payload_copy;
+    job.payload_free = free_store_event_payload;
+    job.result_cb = wait_ctx_result_cb;
+    job.result_cb_ctx = &wait_ctx;
+
+    thread_pool_status_t submit_rc = thread_pool_submit_write(&job, NULL);
+    if (submit_rc != THREAD_POOL_STATUS_OK) {
+        free_store_event_payload(payload_copy);
+        wait_ctx_destroy(&wait_ctx);
+        return -1;
+    }
+
+    wait_ctx_wait(&wait_ctx);
+
+    int rc = -1;
+    if (wait_ctx.status == THREAD_POOL_STATUS_OK && wait_ctx.result_data) {
+        thread_pool_store_event_result_t* result = (thread_pool_store_event_result_t*)wait_ctx.result_data;
+        out_result->step_rc = result->step_rc;
+        out_result->extended_errcode = result->extended_errcode;
+        free(result);
+        rc = 0;
+    }
+
+    wait_ctx_destroy(&wait_ctx);
+    return rc;
+}
+
 __attribute__((constructor))
 static void thread_pool_state_init_once(void) {
     pthread_mutex_init(&g_pool.state_mutex, NULL);

src/thread_pool.h

@@ -56,6 +56,45 @@ typedef struct {
     void* wake_loop_ctx;
 } thread_pool_config_t;
 
+typedef struct {
+    char* sql;
+    char** bind_params;
+    int bind_param_count;
+} thread_pool_count_payload_t;
+
+typedef struct {
+    char* sql;
+    char** bind_params;
+    int bind_param_count;
+    int row_limit;
+} thread_pool_req_payload_t;
+
+typedef struct {
+    char* id;
+    char* pubkey;
+    long long created_at;
+    int kind;
+    char* event_type;
+    char* content;
+    char* sig;
+    char* tags_json;
+    char* event_json;
+} thread_pool_store_event_payload_t;
+
+typedef struct {
+    int count;
+} thread_pool_count_result_t;
+
+typedef struct {
+    char** event_json_rows;
+    int row_count;
+} thread_pool_req_result_t;
+
+typedef struct {
+    int step_rc;
+    int extended_errcode;
+} thread_pool_store_event_result_t;
+
 int thread_pool_init(const thread_pool_config_t* config);
 void thread_pool_shutdown(void);
 int thread_pool_is_running(void);
@@ -63,6 +102,13 @@ int thread_pool_is_running(void);
 thread_pool_status_t thread_pool_submit_read(const thread_pool_job_t* job, uint64_t* out_job_id);
 thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint64_t* out_job_id);
 
+int thread_pool_execute_count_sync(const char* sql, const char** bind_params, int bind_param_count, int* out_count);
+int thread_pool_execute_req_sync(const char* sql, const char** bind_params, int bind_param_count,
+                                 int row_limit, thread_pool_req_result_t** out_result);
+void thread_pool_free_req_result(thread_pool_req_result_t* result);
+int thread_pool_execute_store_event_sync(const thread_pool_store_event_payload_t* payload,
+                                         thread_pool_store_event_result_t* out_result);
+
 #ifdef __cplusplus
 }
 #endif

src/websockets.c

@@ -2867,10 +2867,6 @@ int process_dm_stats_command(cJSON* dm_event, char* error_message, size_t error_
 int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) {
     // pss is now used for query tracking, so remove unused warning suppression
 
-    if (thread_pool_is_running()) {
-        DEBUG_TRACE("Thread pool scaffold active: handle_count_message() still using synchronous DB path");
-    }
-
     if (!cJSON_IsArray(filters)) {
         DEBUG_ERROR("COUNT filters is not an array");
         return 0;
@@ -3132,10 +3128,10 @@ int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, st
         struct timespec query_start, query_end;
         clock_gettime(CLOCK_MONOTONIC, &query_start);
 
-        // Execute count query
+        // Execute count query via thread pool helper
         int filter_count = 0;
-        if (db_count_with_sql(sql, (const char**)bind_params, bind_param_count, &filter_count) != 0) {
+        if (thread_pool_execute_count_sync(sql, (const char**)bind_params, bind_param_count, &filter_count) != 0) {
-            DEBUG_ERROR("Failed to execute COUNT query via db_ops");
+            DEBUG_ERROR("Failed to execute COUNT query via db_ops/thread_pool");
             continue;
         }
         

test_results_20260401_093238.log

@@ -0,0 +1,18 @@
+2026-04-01 09:32:38 - ==========================================
+2026-04-01 09:32:38 - C-Relay Comprehensive Test Suite Runner
+2026-04-01 09:32:38 - ==========================================
+2026-04-01 09:32:38 - Relay URL: ws://127.0.0.1:8888
+2026-04-01 09:32:38 - Log file: test_results_20260401_093238.log
+2026-04-01 09:32:38 - Report file: test_report_20260401_093238.html
+2026-04-01 09:32:38 - 
+2026-04-01 09:32:38 - Checking relay status at ws://127.0.0.1:8888...
+2026-04-01 09:32:38 - \033[0;32m✓ Relay HTTP endpoint is accessible\033[0m
+2026-04-01 09:32:38 - 
+2026-04-01 09:32:38 - Starting comprehensive test execution...
+2026-04-01 09:32:38 - 
+2026-04-01 09:32:38 - \033[0;34m=== SECURITY TEST SUITES ===\033[0m
+2026-04-01 09:32:38 - ==========================================
+2026-04-01 09:32:38 - Running Test Suite: SQL Injection Tests
+2026-04-01 09:32:38 - Description: Comprehensive SQL injection vulnerability testing
+2026-04-01 09:32:38 - ==========================================
+2026-04-01 09:32:38 - \033[0;31mERROR: Test script sql_injection_tests.sh not found\033[0m

tests/filter_validation_test.sh

@@ -8,7 +8,7 @@ set -e
 # Configuration
 RELAY_HOST="127.0.0.1"
 RELAY_PORT="8888"
-TEST_TIMEOUT=5
+TEST_TIMEOUT=10
 
 # Colors for output
 RED='\033[0;31m'
@@ -34,7 +34,7 @@ test_websocket_message() {
 
     # Send message via websocat and capture response
     local response
-    response=$(echo "$message" | timeout $TEST_TIMEOUT websocat -B 1048576 ws://$RELAY_HOST:$RELAY_PORT 2>/dev/null || echo 'CONNECTION_FAILED')
+    response=$(echo "$message" | timeout $TEST_TIMEOUT websocat -B 1048576 ws://$RELAY_HOST:$RELAY_PORT 2>/dev/null | head -1 || echo 'CONNECTION_FAILED')
 
     if [[ "$response" == "CONNECTION_FAILED" ]]; then
         echo -e "${RED}FAILED${NC} - Could not connect to relay"
@@ -42,8 +42,8 @@ test_websocket_message() {
         return 1
     fi
 
-    if [[ "$response" == "TIMEOUT" ]]; then
+    if [[ -z "$response" ]]; then
-        echo -e "${RED}FAILED${NC} - Connection timeout"
+        echo -e "${RED}FAILED${NC} - Empty response"
         FAILED_TESTS=$((FAILED_TESTS + 1))
         return 1
     fi
@@ -119,8 +119,9 @@ test_websocket_message "Invalid author type" '["REQ","sub1",{"authors":[123]}]'
 # Test 6: Invalid author hex
 test_websocket_message "Invalid author hex" '["REQ","sub1",{"authors":["invalid-hex"]}]' "error: invalid author hex string"
 
-# Test 7: Too many authors
+# Test 7: Too many authors (exceed MAX_AUTHORS_PER_FILTER with compact payload)
-test_websocket_message "Too many authors" '["REQ","sub1",{"authors":["a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a"]}]' "error: too many authors"
+too_many_authors=$(yes '"a"' | head -n 1001 | paste -sd, -)
+test_websocket_message "Too many authors" "[\"REQ\",\"sub1\",{\"authors\":[${too_many_authors}]}]" "error: too many authors"
 
 echo
 echo "=== Testing IDs Validation ==="
@@ -131,8 +132,9 @@ test_websocket_message "Invalid ID type" '["REQ","sub1",{"ids":[123]}]' "error:
 # Test 9: Invalid ID hex
 test_websocket_message "Invalid ID hex" '["REQ","sub1",{"ids":["invalid-hex"]}]' "error: invalid id hex string"
 
-# Test 10: Too many IDs
+# Test 10: Too many IDs (exceed MAX_IDS_PER_FILTER with compact payload)
-test_websocket_message "Too many IDs" '["REQ","sub1",{"ids":["a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a","a"]}]' "error: too many ids"
+too_many_ids=$(yes '"a"' | head -n 1001 | paste -sd, -)
+test_websocket_message "Too many IDs" "[\"REQ\",\"sub1\",{\"ids\":[${too_many_ids}]}]" "error: too many ids"
 
 echo
 echo "=== Testing Kinds Validation ==="
@@ -146,8 +148,9 @@ test_websocket_message "Negative kind" '["REQ","sub1",{"kinds":[-1]}]' "error: i
 # Test 13: Too large kind
 test_websocket_message "Too large kind" '["REQ","sub1",{"kinds":[70000]}]' "error: invalid kind value"
 
-# Test 14: Too many kinds
+# Test 14: Too many kinds (exceed MAX_KINDS_PER_FILTER=500)
-test_websocket_message "Too many kinds" '["REQ","sub1",{"kinds":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52]}]' "error: too many kinds"
+too_many_kinds=$(seq -s, 1 501)
+test_websocket_message "Too many kinds" "[\"REQ\",\"sub1\",{\"kinds\":[${too_many_kinds}]}]" "error: too many kinds"
 
 echo
 echo "=== Testing Timestamp Validation ==="
@@ -194,8 +197,9 @@ echo "=== Testing Tag Filter Validation ==="
 # Test 25: Invalid tag filter type
 test_websocket_message "Invalid tag filter type" '["REQ","sub1",{"#e":"not-an-array"}]' "error: #e must be an array"
 
-# Test 26: Too many tag values
+# Test 26: Too many tag values (exceed MAX_TAG_VALUES_PER_FILTER=1000)
-test_websocket_message "Too many tag values" '["REQ","sub1",{"#e":['$(printf '"a%.0s",' {1..101})'"a"]}]' "error: too many #e values"
+too_many_tag_values=$(yes '"a"' | head -n 1001 | paste -sd, -)
+test_websocket_message "Too many tag values" "[\"REQ\",\"sub1\",{\"#e\":[${too_many_tag_values}]}]" "error: too many #e values"
 
 # Test 27: Tag value too long
 test_websocket_message "Tag value too long" '["REQ","sub1",{"#e":["'$(printf 'a%.0s' {1..1025})'"]}]' "error: #e value too long"