v0.1.12 - Successfully integrated websocket client, and program connects to relays and publishes kinds 0 and 10002
This commit is contained in:
Your Name
784
src/relay_client.c
Normal file
784
src/relay_client.c
Normal file
@@ -0,0 +1,784 @@
|
||||
/*
|
||||
* Ginxsom Relay Client Implementation
|
||||
*
|
||||
* Manages connections to Nostr relays, publishes events, and subscribes to admin commands.
|
||||
*/
|
||||
|
||||
#include "relay_client.h"
|
||||
#include "../nostr_core_lib/nostr_core/nostr_core.h"
|
||||
#include <sqlite3.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <pthread.h>
|
||||
#include <unistd.h>
|
||||
#include <time.h>
|
||||
|
||||
// Forward declare app_log to avoid including ginxsom.h (which has typedef conflicts)
|
||||
typedef enum {
|
||||
LOG_DEBUG = 0,
|
||||
LOG_INFO = 1,
|
||||
LOG_WARN = 2,
|
||||
LOG_ERROR = 3
|
||||
} log_level_t;
|
||||
|
||||
void app_log(log_level_t level, const char* format, ...);
|
||||
|
||||
// Maximum number of relays to connect to
|
||||
#define MAX_RELAYS 10
|
||||
|
||||
// Reconnection settings
|
||||
#define RECONNECT_DELAY_SECONDS 30
|
||||
#define MAX_RECONNECT_ATTEMPTS 5
|
||||
|
||||
// Global state
|
||||
static struct {
|
||||
int enabled;
|
||||
int initialized;
|
||||
int running;
|
||||
char db_path[512];
|
||||
nostr_relay_pool_t* pool;
|
||||
char** relay_urls;
|
||||
int relay_count;
|
||||
nostr_pool_subscription_t* admin_subscription;
|
||||
pthread_t management_thread;
|
||||
pthread_mutex_t state_mutex;
|
||||
} g_relay_state = {0};
|
||||
|
||||
// External globals from main.c
|
||||
extern char g_blossom_seckey[65];
|
||||
extern char g_blossom_pubkey[65];
|
||||
extern char g_admin_pubkey[65];
|
||||
|
||||
// Forward declarations
|
||||
static void *relay_management_thread(void *arg);
|
||||
static int load_config_from_db(void);
|
||||
static int parse_relay_urls(const char *json_array);
|
||||
static int subscribe_to_admin_commands(void);
|
||||
static void on_publish_response(const char* relay_url, const char* event_id, int success, const char* message, void* user_data);
|
||||
static void on_admin_command_event(cJSON* event, const char* relay_url, void* user_data);
|
||||
static void on_admin_subscription_eose(cJSON** events, int event_count, void* user_data);
|
||||
|
||||
// Initialize relay client system
|
||||
int relay_client_init(const char *db_path) {
|
||||
if (g_relay_state.initialized) {
|
||||
app_log(LOG_WARN, "Relay client already initialized");
|
||||
return 0;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Initializing relay client system...");
|
||||
|
||||
// Store database path
|
||||
strncpy(g_relay_state.db_path, db_path, sizeof(g_relay_state.db_path) - 1);
|
||||
|
||||
// Initialize mutex
|
||||
if (pthread_mutex_init(&g_relay_state.state_mutex, NULL) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to initialize relay state mutex");
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Load configuration from database
|
||||
if (load_config_from_db() != 0) {
|
||||
app_log(LOG_ERROR, "Failed to load relay configuration from database");
|
||||
pthread_mutex_destroy(&g_relay_state.state_mutex);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Create relay pool if enabled
|
||||
if (g_relay_state.enabled) {
|
||||
// Use default reconnection config (don't free - it's a static structure)
|
||||
nostr_pool_reconnect_config_t* config = nostr_pool_reconnect_config_default();
|
||||
g_relay_state.pool = nostr_relay_pool_create(config);
|
||||
if (!g_relay_state.pool) {
|
||||
app_log(LOG_ERROR, "Failed to create relay pool");
|
||||
pthread_mutex_destroy(&g_relay_state.state_mutex);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Add all relays to pool
|
||||
for (int i = 0; i < g_relay_state.relay_count; i++) {
|
||||
if (nostr_relay_pool_add_relay(g_relay_state.pool, g_relay_state.relay_urls[i]) != NOSTR_SUCCESS) {
|
||||
app_log(LOG_WARN, "Failed to add relay to pool: %s", g_relay_state.relay_urls[i]);
|
||||
}
|
||||
}
|
||||
|
||||
// Trigger initial connection attempts by creating a dummy subscription
|
||||
// This forces ensure_relay_connection() to be called for each relay
|
||||
app_log(LOG_INFO, "Initiating relay connections...");
|
||||
cJSON* dummy_filter = cJSON_CreateObject();
|
||||
cJSON* kinds = cJSON_CreateArray();
|
||||
cJSON_AddItemToArray(kinds, cJSON_CreateNumber(0)); // Kind 0 (will match nothing)
|
||||
cJSON_AddItemToObject(dummy_filter, "kinds", kinds);
|
||||
cJSON_AddNumberToObject(dummy_filter, "limit", 0); // Limit 0 = no results
|
||||
|
||||
nostr_pool_subscription_t* dummy_sub = nostr_relay_pool_subscribe(
|
||||
g_relay_state.pool,
|
||||
(const char**)g_relay_state.relay_urls,
|
||||
g_relay_state.relay_count,
|
||||
dummy_filter,
|
||||
NULL, // No event callback
|
||||
NULL, // No EOSE callback
|
||||
NULL, // No user data
|
||||
1, // close_on_eose
|
||||
1, // enable_deduplication
|
||||
NOSTR_POOL_EOSE_FIRST, // result_mode
|
||||
30, // relay_timeout_seconds
|
||||
30 // eose_timeout_seconds
|
||||
);
|
||||
|
||||
cJSON_Delete(dummy_filter);
|
||||
|
||||
// Immediately close the dummy subscription
|
||||
if (dummy_sub) {
|
||||
nostr_pool_subscription_close(dummy_sub);
|
||||
app_log(LOG_INFO, "Connection attempts initiated for %d relays", g_relay_state.relay_count);
|
||||
} else {
|
||||
app_log(LOG_WARN, "Failed to initiate connection attempts");
|
||||
}
|
||||
}
|
||||
|
||||
g_relay_state.initialized = 1;
|
||||
app_log(LOG_INFO, "Relay client initialized (enabled: %d, relays: %d)",
|
||||
g_relay_state.enabled, g_relay_state.relay_count);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Load configuration from database
|
||||
static int load_config_from_db(void) {
|
||||
sqlite3 *db;
|
||||
sqlite3_stmt *stmt;
|
||||
int rc;
|
||||
|
||||
rc = sqlite3_open_v2(g_relay_state.db_path, &db, SQLITE_OPEN_READONLY, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
app_log(LOG_ERROR, "Cannot open database: %s", sqlite3_errmsg(db));
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Load enable_relay_connect
|
||||
const char *sql = "SELECT value FROM config WHERE key = ?";
|
||||
rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
app_log(LOG_ERROR, "Failed to prepare statement: %s", sqlite3_errmsg(db));
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
|
||||
sqlite3_bind_text(stmt, 1, "enable_relay_connect", -1, SQLITE_STATIC);
|
||||
rc = sqlite3_step(stmt);
|
||||
if (rc == SQLITE_ROW) {
|
||||
const char *value = (const char *)sqlite3_column_text(stmt, 0);
|
||||
g_relay_state.enabled = (strcmp(value, "true") == 0 || strcmp(value, "1") == 0);
|
||||
} else {
|
||||
g_relay_state.enabled = 0;
|
||||
}
|
||||
sqlite3_finalize(stmt);
|
||||
|
||||
// If not enabled, skip loading relay URLs
|
||||
if (!g_relay_state.enabled) {
|
||||
sqlite3_close(db);
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Load kind_10002_tags (relay URLs)
|
||||
rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
app_log(LOG_ERROR, "Failed to prepare statement: %s", sqlite3_errmsg(db));
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
|
||||
sqlite3_bind_text(stmt, 1, "kind_10002_tags", -1, SQLITE_STATIC);
|
||||
rc = sqlite3_step(stmt);
|
||||
if (rc == SQLITE_ROW) {
|
||||
const char *json_array = (const char *)sqlite3_column_text(stmt, 0);
|
||||
if (parse_relay_urls(json_array) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to parse relay URLs from config");
|
||||
sqlite3_finalize(stmt);
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
app_log(LOG_WARN, "No relay URLs configured in kind_10002_tags");
|
||||
}
|
||||
sqlite3_finalize(stmt);
|
||||
|
||||
sqlite3_close(db);
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Parse relay URLs from JSON array
|
||||
static int parse_relay_urls(const char *json_array) {
|
||||
cJSON *root = cJSON_Parse(json_array);
|
||||
if (!root || !cJSON_IsArray(root)) {
|
||||
app_log(LOG_ERROR, "Invalid JSON array for relay URLs");
|
||||
if (root) cJSON_Delete(root);
|
||||
return -1;
|
||||
}
|
||||
|
||||
int count = cJSON_GetArraySize(root);
|
||||
if (count > MAX_RELAYS) {
|
||||
app_log(LOG_WARN, "Too many relays configured (%d), limiting to %d", count, MAX_RELAYS);
|
||||
count = MAX_RELAYS;
|
||||
}
|
||||
|
||||
// Allocate relay URLs array
|
||||
g_relay_state.relay_urls = malloc(count * sizeof(char*));
|
||||
if (!g_relay_state.relay_urls) {
|
||||
cJSON_Delete(root);
|
||||
return -1;
|
||||
}
|
||||
|
||||
g_relay_state.relay_count = 0;
|
||||
for (int i = 0; i < count; i++) {
|
||||
cJSON *item = cJSON_GetArrayItem(root, i);
|
||||
if (cJSON_IsString(item) && item->valuestring) {
|
||||
g_relay_state.relay_urls[g_relay_state.relay_count] = strdup(item->valuestring);
|
||||
if (!g_relay_state.relay_urls[g_relay_state.relay_count]) {
|
||||
// Cleanup on failure
|
||||
for (int j = 0; j < g_relay_state.relay_count; j++) {
|
||||
free(g_relay_state.relay_urls[j]);
|
||||
}
|
||||
free(g_relay_state.relay_urls);
|
||||
cJSON_Delete(root);
|
||||
return -1;
|
||||
}
|
||||
g_relay_state.relay_count++;
|
||||
}
|
||||
}
|
||||
|
||||
cJSON_Delete(root);
|
||||
app_log(LOG_INFO, "Parsed %d relay URLs from configuration", g_relay_state.relay_count);
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Start relay connections
|
||||
int relay_client_start(void) {
|
||||
if (!g_relay_state.initialized) {
|
||||
app_log(LOG_ERROR, "Relay client not initialized");
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (!g_relay_state.enabled) {
|
||||
app_log(LOG_INFO, "Relay client disabled in configuration");
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (g_relay_state.running) {
|
||||
app_log(LOG_WARN, "Relay client already running");
|
||||
return 0;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Starting relay client...");
|
||||
|
||||
// Start management thread
|
||||
g_relay_state.running = 1;
|
||||
if (pthread_create(&g_relay_state.management_thread, NULL, relay_management_thread, NULL) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to create relay management thread");
|
||||
g_relay_state.running = 0;
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Relay client started successfully");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Relay management thread
|
||||
static void *relay_management_thread(void *arg) {
|
||||
(void)arg;
|
||||
|
||||
app_log(LOG_INFO, "Relay management thread started");
|
||||
|
||||
// Wait for at least one relay to connect (max 30 seconds)
|
||||
int connected = 0;
|
||||
for (int i = 0; i < 30 && !connected; i++) {
|
||||
sleep(1);
|
||||
|
||||
// Poll to process connection attempts
|
||||
nostr_relay_pool_poll(g_relay_state.pool, 100);
|
||||
|
||||
// Check if any relay is connected
|
||||
for (int j = 0; j < g_relay_state.relay_count; j++) {
|
||||
nostr_pool_relay_status_t status = nostr_relay_pool_get_relay_status(
|
||||
g_relay_state.pool,
|
||||
g_relay_state.relay_urls[j]
|
||||
);
|
||||
if (status == NOSTR_POOL_RELAY_CONNECTED) {
|
||||
connected = 1;
|
||||
app_log(LOG_INFO, "Relay connected: %s", g_relay_state.relay_urls[j]);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!connected) {
|
||||
app_log(LOG_WARN, "No relays connected after 30 seconds, continuing anyway");
|
||||
}
|
||||
|
||||
// Publish initial events
|
||||
relay_client_publish_kind0();
|
||||
relay_client_publish_kind10002();
|
||||
|
||||
// Subscribe to admin commands
|
||||
subscribe_to_admin_commands();
|
||||
|
||||
// Main loop: poll the relay pool for incoming messages
|
||||
while (g_relay_state.running) {
|
||||
// Poll with 1000ms timeout
|
||||
int events_processed = nostr_relay_pool_poll(g_relay_state.pool, 1000);
|
||||
|
||||
if (events_processed < 0) {
|
||||
app_log(LOG_ERROR, "Error polling relay pool");
|
||||
sleep(1);
|
||||
}
|
||||
// Pool handles all connection management, reconnection, and message processing
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Relay management thread stopping");
|
||||
return NULL;
|
||||
}
|
||||
|
||||
// Stop relay connections
|
||||
void relay_client_stop(void) {
|
||||
if (!g_relay_state.running) {
|
||||
return;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Stopping relay client...");
|
||||
|
||||
g_relay_state.running = 0;
|
||||
|
||||
// Wait for management thread to finish
|
||||
pthread_join(g_relay_state.management_thread, NULL);
|
||||
|
||||
// Close admin subscription
|
||||
if (g_relay_state.admin_subscription) {
|
||||
nostr_pool_subscription_close(g_relay_state.admin_subscription);
|
||||
g_relay_state.admin_subscription = NULL;
|
||||
}
|
||||
|
||||
// Destroy relay pool (automatically disconnects all relays)
|
||||
if (g_relay_state.pool) {
|
||||
nostr_relay_pool_destroy(g_relay_state.pool);
|
||||
g_relay_state.pool = NULL;
|
||||
}
|
||||
|
||||
// Free relay URLs
|
||||
if (g_relay_state.relay_urls) {
|
||||
for (int i = 0; i < g_relay_state.relay_count; i++) {
|
||||
free(g_relay_state.relay_urls[i]);
|
||||
}
|
||||
free(g_relay_state.relay_urls);
|
||||
g_relay_state.relay_urls = NULL;
|
||||
}
|
||||
|
||||
pthread_mutex_destroy(&g_relay_state.state_mutex);
|
||||
|
||||
app_log(LOG_INFO, "Relay client stopped");
|
||||
}
|
||||
|
||||
// Check if relay client is enabled
|
||||
int relay_client_is_enabled(void) {
|
||||
return g_relay_state.enabled;
|
||||
}
|
||||
|
||||
// Publish Kind 0 profile event
|
||||
int relay_client_publish_kind0(void) {
|
||||
if (!g_relay_state.enabled || !g_relay_state.running || !g_relay_state.pool) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Publishing Kind 0 profile event...");
|
||||
|
||||
// Load kind_0_content from database
|
||||
sqlite3 *db;
|
||||
sqlite3_stmt *stmt;
|
||||
int rc;
|
||||
|
||||
rc = sqlite3_open_v2(g_relay_state.db_path, &db, SQLITE_OPEN_READONLY, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
app_log(LOG_ERROR, "Cannot open database: %s", sqlite3_errmsg(db));
|
||||
return -1;
|
||||
}
|
||||
|
||||
const char *sql = "SELECT value FROM config WHERE key = 'kind_0_content'";
|
||||
rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
app_log(LOG_ERROR, "Failed to prepare statement: %s", sqlite3_errmsg(db));
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
|
||||
rc = sqlite3_step(stmt);
|
||||
if (rc != SQLITE_ROW) {
|
||||
app_log(LOG_WARN, "No kind_0_content found in config");
|
||||
sqlite3_finalize(stmt);
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
|
||||
const char *content = (const char *)sqlite3_column_text(stmt, 0);
|
||||
|
||||
// Convert private key from hex to bytes
|
||||
unsigned char privkey_bytes[32];
|
||||
if (nostr_hex_to_bytes(g_blossom_seckey, privkey_bytes, 32) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to convert private key from hex");
|
||||
sqlite3_finalize(stmt);
|
||||
sqlite3_close(db);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Create and sign Kind 0 event using nostr_core_lib
|
||||
cJSON* event = nostr_create_and_sign_event(
|
||||
0, // kind
|
||||
content, // content
|
||||
NULL, // tags (empty for Kind 0)
|
||||
privkey_bytes, // private key
|
||||
time(NULL) // created_at
|
||||
);
|
||||
|
||||
sqlite3_finalize(stmt);
|
||||
sqlite3_close(db);
|
||||
|
||||
if (!event) {
|
||||
app_log(LOG_ERROR, "Failed to create Kind 0 event");
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Publish to all relays using async pool API
|
||||
int result = nostr_relay_pool_publish_async(
|
||||
g_relay_state.pool,
|
||||
(const char**)g_relay_state.relay_urls,
|
||||
g_relay_state.relay_count,
|
||||
event,
|
||||
on_publish_response,
|
||||
(void*)"Kind 0" // user_data to identify event type
|
||||
);
|
||||
|
||||
cJSON_Delete(event);
|
||||
|
||||
if (result == 0) {
|
||||
app_log(LOG_INFO, "Kind 0 profile event publish initiated");
|
||||
return 0;
|
||||
} else {
|
||||
app_log(LOG_ERROR, "Failed to initiate Kind 0 profile event publish");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// Publish Kind 10002 relay list event
|
||||
int relay_client_publish_kind10002(void) {
|
||||
if (!g_relay_state.enabled || !g_relay_state.running || !g_relay_state.pool) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Publishing Kind 10002 relay list event...");
|
||||
|
||||
// Build tags array from configured relays
|
||||
cJSON* tags = cJSON_CreateArray();
|
||||
for (int i = 0; i < g_relay_state.relay_count; i++) {
|
||||
cJSON* tag = cJSON_CreateArray();
|
||||
cJSON_AddItemToArray(tag, cJSON_CreateString("r"));
|
||||
cJSON_AddItemToArray(tag, cJSON_CreateString(g_relay_state.relay_urls[i]));
|
||||
cJSON_AddItemToArray(tags, tag);
|
||||
}
|
||||
|
||||
// Convert private key from hex to bytes
|
||||
unsigned char privkey_bytes[32];
|
||||
if (nostr_hex_to_bytes(g_blossom_seckey, privkey_bytes, 32) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to convert private key from hex");
|
||||
cJSON_Delete(tags);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Create and sign Kind 10002 event
|
||||
cJSON* event = nostr_create_and_sign_event(
|
||||
10002, // kind
|
||||
"", // content (empty for Kind 10002)
|
||||
tags, // tags
|
||||
privkey_bytes, // private key
|
||||
time(NULL) // created_at
|
||||
);
|
||||
|
||||
cJSON_Delete(tags);
|
||||
|
||||
if (!event) {
|
||||
app_log(LOG_ERROR, "Failed to create Kind 10002 event");
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Publish to all relays using async pool API
|
||||
int result = nostr_relay_pool_publish_async(
|
||||
g_relay_state.pool,
|
||||
(const char**)g_relay_state.relay_urls,
|
||||
g_relay_state.relay_count,
|
||||
event,
|
||||
on_publish_response,
|
||||
(void*)"Kind 10002" // user_data to identify event type
|
||||
);
|
||||
|
||||
cJSON_Delete(event);
|
||||
|
||||
if (result == 0) {
|
||||
app_log(LOG_INFO, "Kind 10002 relay list event publish initiated");
|
||||
return 0;
|
||||
} else {
|
||||
app_log(LOG_ERROR, "Failed to initiate Kind 10002 relay list event publish");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// Send Kind 23457 admin response event
|
||||
int relay_client_send_admin_response(const char *recipient_pubkey, const char *response_content) {
|
||||
if (!g_relay_state.enabled || !g_relay_state.running || !g_relay_state.pool) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (!recipient_pubkey || !response_content) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Sending Kind 23457 admin response to %s", recipient_pubkey);
|
||||
|
||||
// TODO: Encrypt response_content using NIP-44
|
||||
// For now, use plaintext (stub implementation)
|
||||
const char *encrypted_content = response_content;
|
||||
|
||||
// Build tags array
|
||||
cJSON* tags = cJSON_CreateArray();
|
||||
cJSON* p_tag = cJSON_CreateArray();
|
||||
cJSON_AddItemToArray(p_tag, cJSON_CreateString("p"));
|
||||
cJSON_AddItemToArray(p_tag, cJSON_CreateString(recipient_pubkey));
|
||||
cJSON_AddItemToArray(tags, p_tag);
|
||||
|
||||
// Convert private key from hex to bytes
|
||||
unsigned char privkey_bytes[32];
|
||||
if (nostr_hex_to_bytes(g_blossom_seckey, privkey_bytes, 32) != 0) {
|
||||
app_log(LOG_ERROR, "Failed to convert private key from hex");
|
||||
cJSON_Delete(tags);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Create and sign Kind 23457 event
|
||||
cJSON* event = nostr_create_and_sign_event(
|
||||
23457, // kind
|
||||
encrypted_content, // content
|
||||
tags, // tags
|
||||
privkey_bytes, // private key
|
||||
time(NULL) // created_at
|
||||
);
|
||||
|
||||
cJSON_Delete(tags);
|
||||
|
||||
if (!event) {
|
||||
app_log(LOG_ERROR, "Failed to create Kind 23457 event");
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Publish to all relays using async pool API
|
||||
int result = nostr_relay_pool_publish_async(
|
||||
g_relay_state.pool,
|
||||
(const char**)g_relay_state.relay_urls,
|
||||
g_relay_state.relay_count,
|
||||
event,
|
||||
on_publish_response,
|
||||
(void*)"Kind 23457" // user_data to identify event type
|
||||
);
|
||||
|
||||
cJSON_Delete(event);
|
||||
|
||||
if (result == 0) {
|
||||
app_log(LOG_INFO, "Kind 23457 admin response publish initiated");
|
||||
return 0;
|
||||
} else {
|
||||
app_log(LOG_ERROR, "Failed to initiate Kind 23457 admin response publish");
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// Callback for publish responses
|
||||
static void on_publish_response(const char* relay_url, const char* event_id, int success, const char* message, void* user_data) {
|
||||
const char* event_type = (const char*)user_data;
|
||||
|
||||
if (success) {
|
||||
app_log(LOG_INFO, "%s event published successfully to %s (ID: %s)",
|
||||
event_type, relay_url, event_id);
|
||||
} else {
|
||||
app_log(LOG_WARN, "%s event rejected by %s: %s",
|
||||
event_type, relay_url, message ? message : "unknown error");
|
||||
}
|
||||
}
|
||||
|
||||
// Callback for received Kind 23456 admin command events
|
||||
static void on_admin_command_event(cJSON* event, const char* relay_url, void* user_data) {
|
||||
(void)user_data;
|
||||
|
||||
app_log(LOG_INFO, "Received Kind 23456 admin command from relay: %s", relay_url);
|
||||
|
||||
// Extract event fields
|
||||
cJSON* kind_json = cJSON_GetObjectItem(event, "kind");
|
||||
cJSON* pubkey_json = cJSON_GetObjectItem(event, "pubkey");
|
||||
cJSON* content_json = cJSON_GetObjectItem(event, "content");
|
||||
cJSON* id_json = cJSON_GetObjectItem(event, "id");
|
||||
|
||||
if (!kind_json || !pubkey_json || !content_json || !id_json) {
|
||||
app_log(LOG_ERROR, "Invalid event structure");
|
||||
return;
|
||||
}
|
||||
|
||||
int kind = cJSON_GetNumberValue(kind_json);
|
||||
const char* sender_pubkey = cJSON_GetStringValue(pubkey_json);
|
||||
const char* encrypted_content = cJSON_GetStringValue(content_json);
|
||||
const char* event_id = cJSON_GetStringValue(id_json);
|
||||
|
||||
if (kind != 23456) {
|
||||
app_log(LOG_WARN, "Unexpected event kind: %d", kind);
|
||||
return;
|
||||
}
|
||||
|
||||
// Verify sender is admin
|
||||
if (strcmp(sender_pubkey, g_admin_pubkey) != 0) {
|
||||
app_log(LOG_WARN, "Ignoring command from non-admin pubkey: %s", sender_pubkey);
|
||||
return;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Processing admin command (event ID: %s)", event_id);
|
||||
|
||||
// TODO: Decrypt content using NIP-44
|
||||
// For now, log the encrypted content
|
||||
app_log(LOG_DEBUG, "Encrypted command content: %s", encrypted_content);
|
||||
|
||||
// TODO: Parse and execute command
|
||||
// TODO: Send response using relay_client_send_admin_response()
|
||||
}
|
||||
|
||||
// Callback for EOSE (End Of Stored Events) - new signature
|
||||
static void on_admin_subscription_eose(cJSON** events, int event_count, void* user_data) {
|
||||
(void)events;
|
||||
(void)event_count;
|
||||
(void)user_data;
|
||||
app_log(LOG_INFO, "Received EOSE for admin command subscription");
|
||||
}
|
||||
|
||||
// Subscribe to admin commands (Kind 23456)
|
||||
static int subscribe_to_admin_commands(void) {
|
||||
if (!g_relay_state.pool) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Subscribing to Kind 23456 admin commands...");
|
||||
|
||||
// Create subscription filter for Kind 23456 events addressed to us
|
||||
cJSON* filter = cJSON_CreateObject();
|
||||
cJSON* kinds = cJSON_CreateArray();
|
||||
cJSON_AddItemToArray(kinds, cJSON_CreateNumber(23456));
|
||||
cJSON_AddItemToObject(filter, "kinds", kinds);
|
||||
|
||||
cJSON* p_tags = cJSON_CreateArray();
|
||||
cJSON_AddItemToArray(p_tags, cJSON_CreateString(g_blossom_pubkey));
|
||||
cJSON_AddItemToObject(filter, "#p", p_tags);
|
||||
|
||||
cJSON_AddNumberToObject(filter, "since", (double)time(NULL));
|
||||
|
||||
// Subscribe using pool with new API signature
|
||||
g_relay_state.admin_subscription = nostr_relay_pool_subscribe(
|
||||
g_relay_state.pool,
|
||||
(const char**)g_relay_state.relay_urls,
|
||||
g_relay_state.relay_count,
|
||||
filter,
|
||||
on_admin_command_event,
|
||||
on_admin_subscription_eose,
|
||||
NULL, // user_data
|
||||
0, // close_on_eose (keep subscription open)
|
||||
1, // enable_deduplication
|
||||
NOSTR_POOL_EOSE_FULL_SET, // result_mode
|
||||
30, // relay_timeout_seconds
|
||||
30 // eose_timeout_seconds
|
||||
);
|
||||
|
||||
cJSON_Delete(filter);
|
||||
|
||||
if (!g_relay_state.admin_subscription) {
|
||||
app_log(LOG_ERROR, "Failed to create admin command subscription");
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Successfully subscribed to admin commands");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Get current relay connection status
|
||||
char *relay_client_get_status(void) {
|
||||
if (!g_relay_state.pool) {
|
||||
return strdup("[]");
|
||||
}
|
||||
|
||||
cJSON *root = cJSON_CreateArray();
|
||||
|
||||
pthread_mutex_lock(&g_relay_state.state_mutex);
|
||||
for (int i = 0; i < g_relay_state.relay_count; i++) {
|
||||
cJSON *relay_obj = cJSON_CreateObject();
|
||||
cJSON_AddStringToObject(relay_obj, "url", g_relay_state.relay_urls[i]);
|
||||
|
||||
// Get status from pool
|
||||
nostr_pool_relay_status_t status = nostr_relay_pool_get_relay_status(
|
||||
g_relay_state.pool,
|
||||
g_relay_state.relay_urls[i]
|
||||
);
|
||||
|
||||
const char *state_str;
|
||||
switch (status) {
|
||||
case NOSTR_POOL_RELAY_CONNECTED: state_str = "connected"; break;
|
||||
case NOSTR_POOL_RELAY_CONNECTING: state_str = "connecting"; break;
|
||||
case NOSTR_POOL_RELAY_ERROR: state_str = "error"; break;
|
||||
default: state_str = "disconnected"; break;
|
||||
}
|
||||
cJSON_AddStringToObject(relay_obj, "state", state_str);
|
||||
|
||||
// Get statistics from pool
|
||||
const nostr_relay_stats_t* stats = nostr_relay_pool_get_relay_stats(
|
||||
g_relay_state.pool,
|
||||
g_relay_state.relay_urls[i]
|
||||
);
|
||||
|
||||
if (stats) {
|
||||
cJSON_AddNumberToObject(relay_obj, "events_received", stats->events_received);
|
||||
cJSON_AddNumberToObject(relay_obj, "events_published", stats->events_published);
|
||||
cJSON_AddNumberToObject(relay_obj, "connection_attempts", stats->connection_attempts);
|
||||
cJSON_AddNumberToObject(relay_obj, "connection_failures", stats->connection_failures);
|
||||
|
||||
if (stats->query_latency_avg > 0) {
|
||||
cJSON_AddNumberToObject(relay_obj, "query_latency_ms", stats->query_latency_avg);
|
||||
}
|
||||
}
|
||||
|
||||
cJSON_AddItemToArray(root, relay_obj);
|
||||
}
|
||||
pthread_mutex_unlock(&g_relay_state.state_mutex);
|
||||
|
||||
char *json_str = cJSON_PrintUnformatted(root);
|
||||
cJSON_Delete(root);
|
||||
|
||||
return json_str;
|
||||
}
|
||||
|
||||
// Force reconnection to all relays
|
||||
int relay_client_reconnect(void) {
|
||||
if (!g_relay_state.enabled || !g_relay_state.running || !g_relay_state.pool) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
app_log(LOG_INFO, "Forcing reconnection to all relays...");
|
||||
|
||||
// Remove and re-add all relays to force reconnection
|
||||
pthread_mutex_lock(&g_relay_state.state_mutex);
|
||||
for (int i = 0; i < g_relay_state.relay_count; i++) {
|
||||
nostr_relay_pool_remove_relay(g_relay_state.pool, g_relay_state.relay_urls[i]);
|
||||
nostr_relay_pool_add_relay(g_relay_state.pool, g_relay_state.relay_urls[i]);
|
||||
}
|
||||
pthread_mutex_unlock(&g_relay_state.state_mutex);
|
||||
|
||||
app_log(LOG_INFO, "Reconnection initiated for all relays");
|
||||
return 0;
|
||||
}
|
||||
Reference in New Issue
Block a user