upleb.uk

Public git repos — served from a NIP-34 GRASP relay at git.upleb.uk

summaryrefslogtreecommitdiff
path: root/components/wisp_relay
diff options
context:
space:
mode:
Diffstat (limited to 'components/wisp_relay')
-rw-r--r--components/wisp_relay/CMakeLists.txt16
-rw-r--r--components/wisp_relay/broadcaster.c33
-rw-r--r--components/wisp_relay/broadcaster.h11
-rw-r--r--components/wisp_relay/deletion.c190
-rw-r--r--components/wisp_relay/deletion.h11
-rw-r--r--components/wisp_relay/flash_monitor.c30
-rw-r--r--components/wisp_relay/flash_monitor.h16
-rw-r--r--components/wisp_relay/handlers.c203
-rw-r--r--components/wisp_relay/handlers.h10
-rw-r--r--components/wisp_relay/idf_component.yml1
-rw-r--r--components/wisp_relay/nip11_relay.c53
-rw-r--r--components/wisp_relay/nip11_relay.h9
-rw-r--r--components/wisp_relay/rate_limiter.c98
-rw-r--r--components/wisp_relay/rate_limiter.h40
-rw-r--r--components/wisp_relay/relay_core.h27
-rw-r--r--components/wisp_relay/relay_types.c21
-rw-r--r--components/wisp_relay/relay_types.h43
-rw-r--r--components/wisp_relay/relay_validator.c176
-rw-r--r--components/wisp_relay/relay_validator.h45
-rw-r--r--components/wisp_relay/router.c140
-rw-r--r--components/wisp_relay/router.h19
-rw-r--r--components/wisp_relay/storage_engine.c402
-rw-r--r--components/wisp_relay/storage_engine.h88
-rw-r--r--components/wisp_relay/sub_manager.c272
-rw-r--r--components/wisp_relay/sub_manager.h92
-rw-r--r--components/wisp_relay/ws_server.c426
-rw-r--r--components/wisp_relay/ws_server.h41
27 files changed, 2513 insertions, 0 deletions
diff --git a/components/wisp_relay/CMakeLists.txt b/components/wisp_relay/CMakeLists.txt
new file mode 100644
index 0000000..5da9a9c
--- /dev/null
+++ b/components/wisp_relay/CMakeLists.txt
@@ -0,0 +1,16 @@
1idf_component_register(
2 SRCS "ws_server.c"
3 "storage_engine.c"
4 "sub_manager.c"
5 "broadcaster.c"
6 "rate_limiter.c"
7 "nip11_relay.c"
8 "deletion.c"
9 "flash_monitor.c"
10 "relay_validator.c"
11 "router.c"
12 "handlers.c"
13 "relay_types.c"
14 INCLUDE_DIRS "."
15 REQUIRES esp_http_server esp_timer nvs_flash log json esp_littlefs mbedtls secp256k1
16)
diff --git a/components/wisp_relay/broadcaster.c b/components/wisp_relay/broadcaster.c
new file mode 100644
index 0000000..738cbdb
--- /dev/null
+++ b/components/wisp_relay/broadcaster.c
@@ -0,0 +1,33 @@
1#include "broadcaster.h"
2#include "relay_core.h"
3#include "router.h"
4#include "sub_manager.h"
5#include "esp_log.h"
6
7static const char *TAG = "broadcaster";
8
9void broadcaster_fanout_json(relay_ctx_t *ctx, const char *event_json,
10 size_t event_len, int event_kind,
11 const char *event_pubkey_hex,
12 uint64_t event_created_at)
13{
14 if (!ctx || !ctx->sub_manager) return;
15
16 sub_match_result_t matches;
17 sub_manager_match_json(ctx->sub_manager, event_json, event_len, event_kind,
18 event_pubkey_hex, event_created_at, &matches);
19
20 if (matches.count == 0) {
21 ESP_LOGD(TAG, "No subscribers for event kind=%d", event_kind);
22 return;
23 }
24
25 ESP_LOGD(TAG, "Broadcasting event kind=%d to %d subscriptions",
26 event_kind, matches.count);
27
28 for (uint8_t i = 0; i < matches.count; i++) {
29 sub_match_entry_t *entry = &matches.matches[i];
30 router_send_event(ctx, entry->conn_fd, entry->sub_id,
31 event_json, event_len);
32 }
33}
diff --git a/components/wisp_relay/broadcaster.h b/components/wisp_relay/broadcaster.h
new file mode 100644
index 0000000..0b29f71
--- /dev/null
+++ b/components/wisp_relay/broadcaster.h
@@ -0,0 +1,11 @@
1#ifndef BROADCASTER_H
2#define BROADCASTER_H
3
4#include "relay_core.h"
5
6void broadcaster_fanout_json(relay_ctx_t *ctx, const char *event_json,
7 size_t event_len, int event_kind,
8 const char *event_pubkey_hex,
9 uint64_t event_created_at);
10
11#endif
diff --git a/components/wisp_relay/deletion.c b/components/wisp_relay/deletion.c
new file mode 100644
index 0000000..7ad3c22
--- /dev/null
+++ b/components/wisp_relay/deletion.c
@@ -0,0 +1,190 @@
1#include "deletion.h"
2#include "relay_types.h"
3#include "cJSON.h"
4#include "esp_log.h"
5#include <inttypes.h>
6#include <stdlib.h>
7#include <string.h>
8
9static const char *TAG = "deletion";
10
11static int extract_event_id_field(const char *event_json, size_t len,
12 uint8_t id_out[32])
13{
14 cJSON *obj = cJSON_ParseWithLength(event_json, len);
15 if (!obj) return -1;
16 cJSON *id_item = cJSON_GetObjectItem(obj, "id");
17 if (!id_item || !cJSON_IsString(id_item) || strlen(id_item->valuestring) != 64) {
18 cJSON_Delete(obj);
19 return -1;
20 }
21 int ret = relay_hex_to_bytes(id_item->valuestring, 64, id_out, 32);
22 cJSON_Delete(obj);
23 return ret;
24}
25
26static char *extract_pubkey_hex(const char *event_json, size_t len)
27{
28 cJSON *obj = cJSON_ParseWithLength(event_json, len);
29 if (!obj) return NULL;
30 cJSON *pk = cJSON_GetObjectItem(obj, "pubkey");
31 char *result = NULL;
32 if (pk && cJSON_IsString(pk)) result = strdup(pk->valuestring);
33 cJSON_Delete(obj);
34 return result;
35}
36
37static int delete_by_e_tags(storage_engine_t *storage, const char *event_json,
38 size_t len, const char *deleter_pubkey)
39{
40 cJSON *obj = cJSON_ParseWithLength(event_json, len);
41 if (!obj) return 0;
42
43 cJSON *tags = cJSON_GetObjectItem(obj, "tags");
44 if (!tags || !cJSON_IsArray(tags)) { cJSON_Delete(obj); return 0; }
45
46 int deleted = 0;
47 int array_size = cJSON_GetArraySize(tags);
48
49 for (int i = 0; i < array_size; i++) {
50 cJSON *tag = cJSON_GetArrayItem(tags, i);
51 if (!cJSON_IsArray(tag)) continue;
52 cJSON *tag_name = cJSON_GetArrayItem(tag, 0);
53 if (!tag_name || !cJSON_IsString(tag_name)) continue;
54 if (strcmp(tag_name->valuestring, "e") != 0) continue;
55
56 cJSON *tag_val = cJSON_GetArrayItem(tag, 1);
57 if (!tag_val || !cJSON_IsString(tag_val)) continue;
58
59 uint8_t event_id[32];
60 if (relay_hex_to_bytes(tag_val->valuestring, 64, event_id, 32) != 0) continue;
61
62 storage_error_t err = storage_delete_event(storage, event_id);
63 if (err == STORAGE_OK) {
64 deleted++;
65 ESP_LOGI(TAG, "Deleted event: %.16s...", tag_val->valuestring);
66 }
67 }
68
69 cJSON_Delete(obj);
70 return deleted;
71}
72
73static int delete_by_a_tags(storage_engine_t *storage, const char *event_json,
74 size_t len, const char *deleter_pubkey,
75 uint64_t created_at)
76{
77 cJSON *obj = cJSON_ParseWithLength(event_json, len);
78 if (!obj) return 0;
79
80 cJSON *tags = cJSON_GetObjectItem(obj, "tags");
81 if (!tags || !cJSON_IsArray(tags)) { cJSON_Delete(obj); return 0; }
82
83 int deleted = 0;
84 int array_size = cJSON_GetArraySize(tags);
85
86 for (int i = 0; i < array_size; i++) {
87 cJSON *tag = cJSON_GetArrayItem(tags, i);
88 if (!cJSON_IsArray(tag)) continue;
89 cJSON *tag_name = cJSON_GetArrayItem(tag, 0);
90 if (!tag_name || !cJSON_IsString(tag_name)) continue;
91 if (strcmp(tag_name->valuestring, "a") != 0) continue;
92
93 cJSON *tag_val = cJSON_GetArrayItem(tag, 1);
94 if (!tag_val || !cJSON_IsString(tag_val)) continue;
95
96 int32_t kind;
97 char pubkey[65] = {0};
98 char d_tag[256] = "";
99 if (sscanf(tag_val->valuestring, "%" SCNd32 ":%64[^:]:%255s",
100 &kind, pubkey, d_tag) < 2)
101 continue;
102
103 if (strcmp(pubkey, deleter_pubkey) != 0) continue;
104
105 char **results = NULL;
106 uint16_t count = 0;
107 storage_query_events_json(storage, kind, pubkey, 100, &results, &count);
108 for (uint16_t e = 0; e < count; e++) {
109 if (storage_delete_event(storage, (const uint8_t *)results[e]) == STORAGE_OK) {
110 deleted++;
111 }
112 }
113 storage_free_query_results(results, count);
114 }
115
116 cJSON_Delete(obj);
117 return deleted;
118}
119
120static int delete_by_k_tags(storage_engine_t *storage, const char *event_json,
121 size_t len, const char *deleter_pubkey,
122 uint64_t created_at)
123{
124 cJSON *obj = cJSON_ParseWithLength(event_json, len);
125 if (!obj) return 0;
126
127 cJSON *tags = cJSON_GetObjectItem(obj, "tags");
128 if (!tags || !cJSON_IsArray(tags)) { cJSON_Delete(obj); return 0; }
129
130 int deleted = 0;
131 int array_size = cJSON_GetArraySize(tags);
132
133 for (int i = 0; i < array_size; i++) {
134 cJSON *tag = cJSON_GetArrayItem(tags, i);
135 if (!cJSON_IsArray(tag)) continue;
136 cJSON *tag_name = cJSON_GetArrayItem(tag, 0);
137 if (!tag_name || !cJSON_IsString(tag_name)) continue;
138 if (strcmp(tag_name->valuestring, "k") != 0) continue;
139
140 cJSON *tag_val = cJSON_GetArrayItem(tag, 1);
141 if (!tag_val || !cJSON_IsString(tag_val)) continue;
142
143 int kind = atoi(tag_val->valuestring);
144
145 char **results = NULL;
146 uint16_t count = 0;
147 storage_query_events_json(storage, kind, deleter_pubkey, 500, &results, &count);
148 for (uint16_t e = 0; e < count; e++) {
149 uint8_t eid[32];
150 if (extract_event_id_field(results[e], strlen(results[e]), eid) == 0) {
151 storage_delete_event(storage, eid);
152 deleted++;
153 }
154 }
155 storage_free_query_results(results, count);
156 }
157
158 cJSON_Delete(obj);
159 return deleted;
160}
161
162int deletion_process_json(storage_engine_t *storage, const char *event_json,
163 size_t event_len)
164{
165 if (!storage || !event_json) return 0;
166
167 cJSON *obj = cJSON_ParseWithLength(event_json, event_len);
168 if (!obj) return 0;
169 cJSON *kind_item = cJSON_GetObjectItem(obj, "kind");
170 int kind = kind_item ? kind_item->valueint : 0;
171 cJSON *pk_item = cJSON_GetObjectItem(obj, "pubkey");
172 const char *pubkey = pk_item ? pk_item->valuestring : "";
173 cJSON *ca_item = cJSON_GetObjectItem(obj, "created_at");
174 uint64_t created_at = ca_item ? (uint64_t)ca_item->valuedouble : 0;
175 cJSON_Delete(obj);
176
177 if (kind != NOSTR_KIND_DELETION) return 0;
178
179 char *deleter_pk = strdup(pubkey);
180 if (!deleter_pk) return 0;
181
182 int deleted = 0;
183 deleted += delete_by_e_tags(storage, event_json, event_len, deleter_pk);
184 deleted += delete_by_a_tags(storage, event_json, event_len, deleter_pk, created_at);
185 deleted += delete_by_k_tags(storage, event_json, event_len, deleter_pk, created_at);
186
187 free(deleter_pk);
188 ESP_LOGI(TAG, "Deletion processed: %d events removed", deleted);
189 return deleted;
190}
diff --git a/components/wisp_relay/deletion.h b/components/wisp_relay/deletion.h
new file mode 100644
index 0000000..b494a8e
--- /dev/null
+++ b/components/wisp_relay/deletion.h
@@ -0,0 +1,11 @@
1#ifndef DELETION_H
2#define DELETION_H
3
4#include "storage_engine.h"
5
6#define NOSTR_KIND_DELETION 5
7
8int deletion_process_json(storage_engine_t *storage, const char *event_json,
9 size_t event_len);
10
11#endif
diff --git a/components/wisp_relay/flash_monitor.c b/components/wisp_relay/flash_monitor.c
new file mode 100644
index 0000000..ceb8c3b
--- /dev/null
+++ b/components/wisp_relay/flash_monitor.c
@@ -0,0 +1,30 @@
1#include "flash_monitor.h"
2#include "esp_littlefs.h"
3#include "esp_log.h"
4#include <string.h>
5
6static const char *TAG = "flash_monitor";
7
8void flash_get_health(const char *partition_label, flash_health_t *health)
9{
10 memset(health, 0, sizeof(flash_health_t));
11
12 esp_err_t ret = esp_littlefs_info(partition_label,
13 &health->total_bytes,
14 &health->used_bytes);
15 if (ret != ESP_OK) {
16 ESP_LOGE(TAG, "Failed to get LittleFS info: %s", esp_err_to_name(ret));
17 return;
18 }
19
20 if (health->total_bytes == 0) {
21 health->free_bytes = 0;
22 health->usage_percent = 0.0f;
23 } else {
24 health->free_bytes = health->total_bytes - health->used_bytes;
25 health->usage_percent = (float)health->used_bytes / health->total_bytes * 100.0f;
26 }
27
28 ESP_LOGD(TAG, "Flash: %.1f%% used (%zu/%zu bytes)",
29 health->usage_percent, health->used_bytes, health->total_bytes);
30}
diff --git a/components/wisp_relay/flash_monitor.h b/components/wisp_relay/flash_monitor.h
new file mode 100644
index 0000000..86f1b53
--- /dev/null
+++ b/components/wisp_relay/flash_monitor.h
@@ -0,0 +1,16 @@
1#ifndef FLASH_MONITOR_H
2#define FLASH_MONITOR_H
3
4#include <stdint.h>
5#include <stddef.h>
6
7typedef struct {
8 size_t total_bytes;
9 size_t used_bytes;
10 size_t free_bytes;
11 float usage_percent;
12} flash_health_t;
13
14void flash_get_health(const char *partition_label, flash_health_t *health);
15
16#endif
diff --git a/components/wisp_relay/handlers.c b/components/wisp_relay/handlers.c
new file mode 100644
index 0000000..2164725
--- /dev/null
+++ b/components/wisp_relay/handlers.c
@@ -0,0 +1,203 @@
1#include "handlers.h"
2#include "router.h"
3#include "storage_engine.h"
4#include "sub_manager.h"
5#include "relay_validator.h"
6#include "broadcaster.h"
7#include "deletion.h"
8#include "rate_limiter.h"
9#include "relay_types.h"
10#include "cJSON.h"
11#include "esp_log.h"
12#include <string.h>
13
14static const char *TAG = "handlers";
15
16int handle_event(relay_ctx_t *ctx, int conn_fd, const char *event_json, size_t event_len)
17{
18 if (!ctx || !event_json) return -1;
19
20 if (ctx->rate_limiter) {
21 if (!rate_limiter_check(ctx->rate_limiter, conn_fd, RATE_TYPE_EVENT)) {
22 router_send_ok(ctx, conn_fd, "", false, "rate limited");
23 return -1;
24 }
25 }
26
27 cJSON *obj = cJSON_ParseWithLength(event_json, event_len);
28 if (!obj) {
29 router_send_ok(ctx, conn_fd, "", false, "invalid JSON");
30 return -1;
31 }
32
33 cJSON *id_item = cJSON_GetObjectItem(obj, "id");
34 cJSON *pubkey_item = cJSON_GetObjectItem(obj, "pubkey");
35 cJSON *kind_item = cJSON_GetObjectItem(obj, "kind");
36 cJSON *ca_item = cJSON_GetObjectItem(obj, "created_at");
37
38 if (!id_item || !pubkey_item || !kind_item || !ca_item) {
39 cJSON_Delete(obj);
40 router_send_ok(ctx, conn_fd, "", false, "missing required fields");
41 return -1;
42 }
43
44 const char *id_hex = id_item->valuestring;
45 const char *pubkey_hex = pubkey_item->valuestring;
46 int kind = kind_item->valueint;
47 uint64_t created_at = (uint64_t)ca_item->valuedouble;
48
49 if (ctx->config.max_future_sec > 0) {
50 int64_t now = (int64_t)(xTaskGetTickCount() / configTICK_RATE_HZ);
51 if ((int64_t)created_at > now + ctx->config.max_future_sec) {
52 cJSON_Delete(obj);
53 router_send_ok(ctx, conn_fd, id_hex, false, "created_at too far in future");
54 return -1;
55 }
56 }
57
58 uint8_t event_id[32];
59 if (relay_hex_to_bytes(id_hex, 64, event_id, 32) != 0) {
60 cJSON_Delete(obj);
61 router_send_ok(ctx, conn_fd, "", false, "invalid event id");
62 return -1;
63 }
64
65 if (storage_event_exists(ctx->storage, event_id)) {
66 cJSON_Delete(obj);
67 router_send_ok(ctx, conn_fd, id_hex, true, "duplicate");
68 return 0;
69 }
70
71 if (!relay_validator_verify_event(event_json, event_len)) {
72 cJSON_Delete(obj);
73 router_send_ok(ctx, conn_fd, id_hex, false, "invalid signature");
74 return -1;
75 }
76
77 cJSON_Delete(obj);
78
79 storage_error_t err = storage_save_event_json(ctx->storage, event_json, event_len);
80 if (err != STORAGE_OK) {
81 const char *msg = (err == STORAGE_ERR_FULL) ? "relay full" :
82 (err == STORAGE_ERR_DUPLICATE) ? "duplicate" : "storage error";
83 router_send_ok(ctx, conn_fd, id_hex, false, msg);
84 return -1;
85 }
86
87 router_send_ok(ctx, conn_fd, id_hex, true, "");
88
89 if (kind == NOSTR_KIND_DELETION) {
90 deletion_process_json(ctx->storage, event_json, event_len);
91 }
92
93 broadcaster_fanout_json(ctx, event_json, event_len, kind, pubkey_hex, created_at);
94
95 return 0;
96}
97
98static void parse_filter_json(const char *json, sub_filter_t *filter)
99{
100 memset(filter, 0, sizeof(sub_filter_t));
101 cJSON *obj = cJSON_Parse(json);
102 if (!obj) return;
103
104 cJSON *arr;
105
106 arr = cJSON_GetObjectItem(obj, "ids");
107 if (arr && cJSON_IsArray(arr)) {
108 filter->ids_count = cJSON_GetArraySize(arr);
109 if (filter->ids_count > SUB_MAX_FILTER_IDS) filter->ids_count = SUB_MAX_FILTER_IDS;
110 for (size_t i = 0; i < filter->ids_count; i++)
111 filter->ids[i] = strdup(cJSON_GetArrayItem(arr, i)->valuestring);
112 }
113
114 arr = cJSON_GetObjectItem(obj, "authors");
115 if (arr && cJSON_IsArray(arr)) {
116 filter->authors_count = cJSON_GetArraySize(arr);
117 if (filter->authors_count > SUB_MAX_FILTER_AUTHORS) filter->authors_count = SUB_MAX_FILTER_AUTHORS;
118 for (size_t i = 0; i < filter->authors_count; i++)
119 filter->authors[i] = strdup(cJSON_GetArrayItem(arr, i)->valuestring);
120 }
121
122 arr = cJSON_GetObjectItem(obj, "kinds");
123 if (arr && cJSON_IsArray(arr)) {
124 filter->kinds_count = cJSON_GetArraySize(arr);
125 if (filter->kinds_count > SUB_MAX_FILTER_KINDS) filter->kinds_count = SUB_MAX_FILTER_KINDS;
126 for (size_t i = 0; i < filter->kinds_count; i++)
127 filter->kinds[i] = cJSON_GetArrayItem(arr, i)->valueint;
128 }
129
130 arr = cJSON_GetObjectItem(obj, "#e");
131 if (arr && cJSON_IsArray(arr)) {
132 filter->e_tags_count = cJSON_GetArraySize(arr);
133 if (filter->e_tags_count > SUB_MAX_FILTER_ETAGS) filter->e_tags_count = SUB_MAX_FILTER_ETAGS;
134 for (size_t i = 0; i < filter->e_tags_count; i++)
135 filter->e_tags[i] = strdup(cJSON_GetArrayItem(arr, i)->valuestring);
136 }
137
138 arr = cJSON_GetObjectItem(obj, "#p");
139 if (arr && cJSON_IsArray(arr)) {
140 filter->p_tags_count = cJSON_GetArraySize(arr);
141 if (filter->p_tags_count > SUB_MAX_FILTER_PTAGS) filter->p_tags_count = SUB_MAX_FILTER_PTAGS;
142 for (size_t i = 0; i < filter->p_tags_count; i++)
143 filter->p_tags[i] = strdup(cJSON_GetArrayItem(arr, i)->valuestring);
144 }
145
146 cJSON *since = cJSON_GetObjectItem(obj, "since");
147 if (since) filter->since = (int64_t)since->valuedouble;
148 cJSON *until = cJSON_GetObjectItem(obj, "until");
149 if (until) filter->until = (int64_t)until->valuedouble;
150 cJSON *limit = cJSON_GetObjectItem(obj, "limit");
151 if (limit) filter->limit = limit->valueint;
152
153 cJSON_Delete(obj);
154}
155
156void handle_req(relay_ctx_t *ctx, int conn_fd, const char *sub_id, const char *filters_json)
157{
158 if (!ctx || !sub_id) return;
159
160 if (ctx->rate_limiter) {
161 if (!rate_limiter_check(ctx->rate_limiter, conn_fd, RATE_TYPE_REQ)) {
162 router_send_closed(ctx, conn_fd, sub_id, "rate limited");
163 return;
164 }
165 }
166
167 sub_filter_t filter;
168 parse_filter_json(filters_json, &filter);
169
170 int query_kind = -1;
171 const char *query_author = NULL;
172 int query_limit = filter.limit > 0 ? filter.limit : 100;
173
174 if (filter.kinds_count > 0) query_kind = filter.kinds[0];
175 if (filter.authors_count > 0) query_author = filter.authors[0];
176
177 char **results = NULL;
178 uint16_t count = 0;
179 storage_query_events_json(ctx->storage, query_kind, query_author,
180 query_limit, &results, &count);
181
182 for (uint16_t i = 0; i < count; i++) {
183 router_send_event(ctx, conn_fd, sub_id, results[i], strlen(results[i]));
184 }
185 storage_free_query_results(results, count);
186
187 router_send_eose(ctx, conn_fd, sub_id);
188
189 sub_manager_add(ctx->sub_manager, conn_fd, sub_id, &filter, 1);
190
191 sub_filter_t *f = &filter;
192 for (size_t i = 0; i < f->ids_count; i++) free(f->ids[i]);
193 for (size_t i = 0; i < f->authors_count; i++) free(f->authors[i]);
194 for (size_t i = 0; i < f->e_tags_count; i++) free(f->e_tags[i]);
195 for (size_t i = 0; i < f->p_tags_count; i++) free(f->p_tags[i]);
196}
197
198int handle_close(relay_ctx_t *ctx, int conn_fd, const char *sub_id)
199{
200 if (!ctx || !sub_id) return -1;
201 sub_manager_remove(ctx->sub_manager, conn_fd, sub_id);
202 return 0;
203}
diff --git a/components/wisp_relay/handlers.h b/components/wisp_relay/handlers.h
new file mode 100644
index 0000000..91621bf
--- /dev/null
+++ b/components/wisp_relay/handlers.h
@@ -0,0 +1,10 @@
1#ifndef HANDLERS_H
2#define HANDLERS_H
3
4#include "relay_core.h"
5
6int handle_event(relay_ctx_t *ctx, int conn_fd, const char *event_json, size_t event_len);
7void handle_req(relay_ctx_t *ctx, int conn_fd, const char *sub_id, const char *filters_json);
8int handle_close(relay_ctx_t *ctx, int conn_fd, const char *sub_id);
9
10#endif
diff --git a/components/wisp_relay/idf_component.yml b/components/wisp_relay/idf_component.yml
new file mode 100644
index 0000000..c093387
--- /dev/null
+++ b/components/wisp_relay/idf_component.yml
@@ -0,0 +1 @@
dependencies: {}
diff --git a/components/wisp_relay/nip11_relay.c b/components/wisp_relay/nip11_relay.c
new file mode 100644
index 0000000..4e1df37
--- /dev/null
+++ b/components/wisp_relay/nip11_relay.c
@@ -0,0 +1,53 @@
1#include "nip11_relay.h"
2#include <string.h>
3
4static const char *NIP11_JSON =
5"{"
6 "\"name\":\"TollGate Relay\","
7 "\"description\":\"Local Nostr relay with 21-day TTL and negentropy sync\","
8 "\"pubkey\":\"\","
9 "\"contact\":\"\","
10 "\"supported_nips\":[1,9,11,20,40,77],"
11 "\"software\":\"https://github.com/nicobao/esp32-tollgate\","
12 "\"version\":\"1.0.0\","
13 "\"limitation\":{"
14 "\"max_message_length\":65536,"
15 "\"max_subscriptions\":8,"
16 "\"max_filters\":4,"
17 "\"max_limit\":500,"
18 "\"max_subid_length\":64,"
19 "\"max_event_tags\":100,"
20 "\"max_content_length\":32768,"
21 "\"min_pow_difficulty\":0,"
22 "\"auth_required\":false,"
23 "\"payment_required\":false"
24 "},"
25 "\"retention\":[{\"kinds\":[0,1,2,3,4,5,6,7],\"time\":1814400}],"
26 "\"relay_countries\":[\"DE\"]"
27"}";
28
29esp_err_t relay_nip11_handler(httpd_req_t *req)
30{
31 char accept[64] = "";
32 httpd_req_get_hdr_value_str(req, "Accept", accept, sizeof(accept));
33
34 if (strstr(accept, "application/nostr+json")) {
35 httpd_resp_set_type(req, "application/nostr+json");
36 } else {
37 httpd_resp_set_type(req, "application/json");
38 }
39
40 httpd_resp_set_hdr(req, "Access-Control-Allow-Origin", "*");
41 httpd_resp_set_hdr(req, "Access-Control-Allow-Headers", "Content-Type, Accept");
42 httpd_resp_set_hdr(req, "Access-Control-Allow-Methods", "GET, OPTIONS");
43 return httpd_resp_send(req, NIP11_JSON, strlen(NIP11_JSON));
44}
45
46esp_err_t relay_nip11_options_handler(httpd_req_t *req)
47{
48 httpd_resp_set_hdr(req, "Access-Control-Allow-Origin", "*");
49 httpd_resp_set_hdr(req, "Access-Control-Allow-Headers", "Content-Type, Accept");
50 httpd_resp_set_hdr(req, "Access-Control-Allow-Methods", "GET, OPTIONS");
51 httpd_resp_set_status(req, "204 No Content");
52 return httpd_resp_send(req, NULL, 0);
53}
diff --git a/components/wisp_relay/nip11_relay.h b/components/wisp_relay/nip11_relay.h
new file mode 100644
index 0000000..84f7971
--- /dev/null
+++ b/components/wisp_relay/nip11_relay.h
@@ -0,0 +1,9 @@
1#ifndef NIP11_RELAY_H
2#define NIP11_RELAY_H
3
4#include "esp_http_server.h"
5
6esp_err_t relay_nip11_handler(httpd_req_t *req);
7esp_err_t relay_nip11_options_handler(httpd_req_t *req);
8
9#endif
diff --git a/components/wisp_relay/rate_limiter.c b/components/wisp_relay/rate_limiter.c
new file mode 100644
index 0000000..7734e03
--- /dev/null
+++ b/components/wisp_relay/rate_limiter.c
@@ -0,0 +1,98 @@
1#include "rate_limiter.h"
2#include "esp_timer.h"
3#include "esp_log.h"
4#include <string.h>
5
6static const char *TAG = "rate_limiter";
7
8void rate_limiter_init(rate_limiter_t *rl, const rate_config_t *config)
9{
10 memset(rl, 0, sizeof(rate_limiter_t));
11 rl->lock = xSemaphoreCreateMutex();
12 if (config) {
13 memcpy(&rl->config, config, sizeof(rate_config_t));
14 } else {
15 rl->config.events_per_minute = 30;
16 rl->config.reqs_per_minute = 60;
17 }
18}
19
20void rate_limiter_destroy(rate_limiter_t *rl)
21{
22 if (!rl) return;
23 if (rl->lock) {
24 vSemaphoreDelete(rl->lock);
25 rl->lock = NULL;
26 }
27}
28
29static rate_bucket_t* get_bucket(rate_limiter_t *rl, int fd)
30{
31 for (int i = 0; i < RATE_LIMITER_MAX_BUCKETS; i++) {
32 if (rl->buckets[i].active && rl->buckets[i].fd == fd) {
33 return &rl->buckets[i];
34 }
35 }
36 for (int i = 0; i < RATE_LIMITER_MAX_BUCKETS; i++) {
37 if (!rl->buckets[i].active) {
38 rl->buckets[i].fd = fd;
39 rl->buckets[i].active = true;
40 rl->buckets[i].event_count = 0;
41 rl->buckets[i].req_count = 0;
42 rl->buckets[i].window_start = esp_timer_get_time() / 1000000;
43 return &rl->buckets[i];
44 }
45 }
46 return NULL;
47}
48
49bool rate_limiter_check(rate_limiter_t *rl, int fd, rate_type_t type)
50{
51 xSemaphoreTake(rl->lock, portMAX_DELAY);
52
53 rate_bucket_t *bucket = get_bucket(rl, fd);
54 if (!bucket) {
55 xSemaphoreGive(rl->lock);
56 return false;
57 }
58
59 uint32_t now = esp_timer_get_time() / 1000000;
60
61 if (now - bucket->window_start >= 60) {
62 bucket->event_count = 0;
63 bucket->req_count = 0;
64 bucket->window_start = now;
65 }
66
67 bool allowed = true;
68 if (type == RATE_TYPE_EVENT) {
69 if (bucket->event_count >= rl->config.events_per_minute) {
70 ESP_LOGW(TAG, "Rate limited: fd=%d events=%d", fd, bucket->event_count);
71 allowed = false;
72 } else {
73 bucket->event_count++;
74 }
75 } else {
76 if (bucket->req_count >= rl->config.reqs_per_minute) {
77 ESP_LOGW(TAG, "Rate limited: fd=%d reqs=%d", fd, bucket->req_count);
78 allowed = false;
79 } else {
80 bucket->req_count++;
81 }
82 }
83
84 xSemaphoreGive(rl->lock);
85 return allowed;
86}
87
88void rate_limiter_reset(rate_limiter_t *rl, int fd)
89{
90 xSemaphoreTake(rl->lock, portMAX_DELAY);
91 for (int i = 0; i < RATE_LIMITER_MAX_BUCKETS; i++) {
92 if (rl->buckets[i].active && rl->buckets[i].fd == fd) {
93 rl->buckets[i].active = false;
94 break;
95 }
96 }
97 xSemaphoreGive(rl->lock);
98}
diff --git a/components/wisp_relay/rate_limiter.h b/components/wisp_relay/rate_limiter.h
new file mode 100644
index 0000000..655ddf2
--- /dev/null
+++ b/components/wisp_relay/rate_limiter.h
@@ -0,0 +1,40 @@
1#ifndef RATE_LIMITER_H
2#define RATE_LIMITER_H
3
4#include <stdint.h>
5#include <stdbool.h>
6#include "freertos/FreeRTOS.h"
7#include "freertos/semphr.h"
8
9#define RATE_LIMITER_MAX_BUCKETS 16
10
11typedef enum {
12 RATE_TYPE_EVENT,
13 RATE_TYPE_REQ,
14} rate_type_t;
15
16typedef struct {
17 uint16_t events_per_minute;
18 uint16_t reqs_per_minute;
19} rate_config_t;
20
21typedef struct {
22 int fd;
23 uint16_t event_count;
24 uint16_t req_count;
25 uint32_t window_start;
26 bool active;
27} rate_bucket_t;
28
29typedef struct rate_limiter {
30 rate_config_t config;
31 rate_bucket_t buckets[RATE_LIMITER_MAX_BUCKETS];
32 SemaphoreHandle_t lock;
33} rate_limiter_t;
34
35void rate_limiter_init(rate_limiter_t *rl, const rate_config_t *config);
36void rate_limiter_destroy(rate_limiter_t *rl);
37bool rate_limiter_check(rate_limiter_t *rl, int fd, rate_type_t type);
38void rate_limiter_reset(rate_limiter_t *rl, int fd);
39
40#endif
diff --git a/components/wisp_relay/relay_core.h b/components/wisp_relay/relay_core.h
new file mode 100644
index 0000000..d8e7096
--- /dev/null
+++ b/components/wisp_relay/relay_core.h
@@ -0,0 +1,27 @@
1#ifndef RELAY_CORE_H
2#define RELAY_CORE_H
3
4#include <stdint.h>
5
6#include "ws_server.h"
7
8typedef struct sub_manager sub_manager_t;
9typedef struct storage_engine storage_engine_t;
10typedef struct rate_limiter rate_limiter_t;
11
12typedef struct relay_ctx {
13 ws_server_t ws_server;
14 sub_manager_t *sub_manager;
15 storage_engine_t *storage;
16 rate_limiter_t *rate_limiter;
17
18 struct {
19 uint16_t port;
20 uint32_t max_event_age_sec;
21 uint8_t max_subs_per_conn;
22 uint8_t max_filters_per_sub;
23 int64_t max_future_sec;
24 } config;
25} relay_ctx_t;
26
27#endif
diff --git a/components/wisp_relay/relay_types.c b/components/wisp_relay/relay_types.c
new file mode 100644
index 0000000..9833885
--- /dev/null
+++ b/components/wisp_relay/relay_types.c
@@ -0,0 +1,21 @@
1#include "relay_types.h"
2#include <stdio.h>
3#include <string.h>
4
5int relay_hex_to_bytes(const char *hex, size_t hex_len, uint8_t *out, size_t out_len)
6{
7 if (hex_len != out_len * 2) return -1;
8 for (size_t i = 0; i < out_len; i++) {
9 unsigned int byte;
10 if (sscanf(hex + i * 2, "%02x", &byte) != 1) return -1;
11 out[i] = (uint8_t)byte;
12 }
13 return 0;
14}
15
16void relay_bytes_to_hex(const uint8_t *bytes, size_t len, char *hex)
17{
18 for (size_t i = 0; i < len; i++)
19 sprintf(hex + i * 2, "%02x", bytes[i]);
20 hex[len * 2] = '\0';
21}
diff --git a/components/wisp_relay/relay_types.h b/components/wisp_relay/relay_types.h
new file mode 100644
index 0000000..343e51b
--- /dev/null
+++ b/components/wisp_relay/relay_types.h
@@ -0,0 +1,43 @@
1#ifndef RELAY_TYPES_H
2#define RELAY_TYPES_H
3
4#include <stdint.h>
5#include <stdbool.h>
6#include <stddef.h>
7
8#define RELAY_MAX_EVENT_SIZE 8192
9#define RELAY_ID_SIZE 32
10#define RELAY_SIG_SIZE 64
11#define RELAY_MAX_TAGS 100
12#define RELAY_MAX_TAG_VALUES 10
13
14typedef struct relay_event {
15 uint8_t id[RELAY_ID_SIZE];
16 uint8_t pubkey[RELAY_ID_SIZE];
17 uint64_t created_at;
18 int kind;
19 uint8_t sig[RELAY_SIG_SIZE];
20 char content[RELAY_MAX_EVENT_SIZE];
21 size_t content_len;
22} relay_event_t;
23
24typedef struct {
25 char **ids;
26 size_t ids_count;
27 char **authors;
28 size_t authors_count;
29 int32_t *kinds;
30 size_t kinds_count;
31 char **e_tags;
32 size_t e_tags_count;
33 char **p_tags;
34 size_t p_tags_count;
35 int64_t since;
36 int64_t until;
37 int limit;
38} relay_filter_t;
39
40int relay_hex_to_bytes(const char *hex, size_t hex_len, uint8_t *out, size_t out_len);
41void relay_bytes_to_hex(const uint8_t *bytes, size_t len, char *hex);
42
43#endif
diff --git a/components/wisp_relay/relay_validator.c b/components/wisp_relay/relay_validator.c
new file mode 100644
index 0000000..eb40d22
--- /dev/null
+++ b/components/wisp_relay/relay_validator.c
@@ -0,0 +1,176 @@
1#include "relay_validator.h"
2#include "relay_types.h"
3#include "esp_log.h"
4#include "mbedtls/sha256.h"
5#include "secp256k1.h"
6#include "secp256k1_extrakeys.h"
7#include "secp256k1_schnorrsig.h"
8#include "cJSON.h"
9#include "freertos/FreeRTOS.h"
10#include "freertos/task.h"
11#include <stddef.h>
12#include <string.h>
13#include <stdlib.h>
14#include <stdio.h>
15
16static const char *TAG = "relay_validator";
17
18static int hex_to_bytes(const char *hex, size_t hex_len, uint8_t *out, size_t out_len)
19{
20 if (hex_len != out_len * 2) return -1;
21 for (size_t i = 0; i < out_len; i++) {
22 unsigned int byte;
23 if (sscanf(hex + i * 2, "%02x", &byte) != 1) return -1;
24 out[i] = (uint8_t)byte;
25 }
26 return 0;
27}
28
29static char *serialize_event_for_id(const char *event_json, size_t event_len)
30{
31 cJSON *obj = cJSON_ParseWithLength(event_json, event_len);
32 if (!obj) return NULL;
33
34 cJSON *serial = cJSON_CreateArray();
35 cJSON_AddItemToArray(serial, cJSON_CreateNumber(0));
36 cJSON_AddItemToArray(serial, cJSON_CreateString(
37 cJSON_GetObjectItem(obj, "pubkey")->valuestring));
38 cJSON_AddItemToArray(serial, cJSON_CreateNumber(
39 cJSON_GetObjectItem(obj, "created_at")->valuedouble));
40 cJSON_AddItemToArray(serial, cJSON_CreateNumber(
41 cJSON_GetObjectItem(obj, "kind")->valueint));
42 cJSON *tags = cJSON_GetObjectItem(obj, "tags");
43 cJSON_AddItemToArray(serial, cJSON_Duplicate(tags, 1));
44 cJSON_AddItemToArray(serial, cJSON_CreateString(
45 cJSON_GetObjectItem(obj, "content")->valuestring));
46
47 char *result = cJSON_PrintUnformatted(serial);
48 cJSON_Delete(serial);
49 cJSON_Delete(obj);
50 return result;
51}
52
53static bool verify_event_id(const char *event_json, size_t event_len,
54 const uint8_t expected_id[32])
55{
56 char *serialized = serialize_event_for_id(event_json, event_len);
57 if (!serialized) return false;
58
59 uint8_t hash[32];
60 mbedtls_sha256((const unsigned char *)serialized, strlen(serialized), hash, 0);
61 free(serialized);
62
63 return memcmp(hash, expected_id, 32) == 0;
64}
65
66static bool verify_schnorr_sig(const uint8_t pubkey[32], const uint8_t msg[32],
67 const uint8_t sig[64])
68{
69 secp256k1_context *ctx = secp256k1_context_create(SECP256K1_CONTEXT_VERIFY);
70 if (!ctx) return false;
71
72 secp256k1_xonly_pubkey xonly_pub;
73 if (!secp256k1_xonly_pubkey_parse(ctx, &xonly_pub, pubkey)) {
74 secp256k1_context_destroy(ctx);
75 return false;
76 }
77
78 bool valid = secp256k1_schnorrsig_verify(ctx, sig, msg, 32, &xonly_pub);
79 secp256k1_context_destroy(ctx);
80 return valid;
81}
82
83bool relay_validator_verify_event(const char *event_json, size_t event_len)
84{
85 cJSON *obj = cJSON_ParseWithLength(event_json, event_len);
86 if (!obj) {
87 ESP_LOGD(TAG, "Invalid JSON");
88 return false;
89 }
90
91 cJSON *id_item = cJSON_GetObjectItem(obj, "id");
92 cJSON *pk_item = cJSON_GetObjectItem(obj, "pubkey");
93 cJSON *sig_item = cJSON_GetObjectItem(obj, "sig");
94
95 if (!id_item || !pk_item || !sig_item) {
96 cJSON_Delete(obj);
97 ESP_LOGD(TAG, "Missing required fields");
98 return false;
99 }
100
101 const char *id_hex = id_item->valuestring;
102 const char *pk_hex = pk_item->valuestring;
103 const char *sig_hex = sig_item->valuestring;
104
105 if (strlen(id_hex) != 64 || strlen(pk_hex) != 64 || strlen(sig_hex) != 128) {
106 cJSON_Delete(obj);
107 ESP_LOGD(TAG, "Invalid field lengths");
108 return false;
109 }
110
111 uint8_t event_id[32], pubkey[32], sig[64];
112 if (hex_to_bytes(id_hex, 64, event_id, 32) != 0 ||
113 hex_to_bytes(pk_hex, 64, pubkey, 32) != 0 ||
114 hex_to_bytes(sig_hex, 128, sig, 64) != 0) {
115 cJSON_Delete(obj);
116 ESP_LOGD(TAG, "Invalid hex encoding");
117 return false;
118 }
119
120 cJSON_Delete(obj);
121
122 if (!verify_event_id(event_json, event_len, event_id)) {
123 ESP_LOGD(TAG, "Event ID mismatch");
124 return false;
125 }
126
127 if (!verify_schnorr_sig(pubkey, event_id, sig)) {
128 ESP_LOGD(TAG, "Invalid signature");
129 return false;
130 }
131
132 return true;
133}
134
135validation_result_t relay_validator_check(const uint8_t *id,
136 const uint8_t *pubkey,
137 uint64_t created_at,
138 int kind,
139 const char *content,
140 size_t content_len,
141 const char *tags_json,
142 const uint8_t *sig,
143 const validator_config_t *config)
144{
145 (void)content; (void)content_len; (void)tags_json;
146
147 if (config) {
148 if (config->max_future_sec > 0) {
149 int64_t now = (int64_t)(xTaskGetTickCount() / configTICK_RATE_HZ);
150 if ((int64_t)created_at > now + config->max_future_sec)
151 return VALIDATION_ERR_FUTURE;
152 }
153 }
154
155 if (!verify_schnorr_sig(pubkey, id, sig))
156 return VALIDATION_ERR_SIG;
157
158 return VALIDATION_OK;
159}
160
161const char *relay_validator_result_string(validation_result_t result)
162{
163 switch (result) {
164 case VALIDATION_OK: return "ok";
165 case VALIDATION_ERR_SCHEMA: return "invalid: schema";
166 case VALIDATION_ERR_ID: return "invalid: event id";
167 case VALIDATION_ERR_SIG: return "invalid: signature";
168 case VALIDATION_ERR_EXPIRED: return "invalid: expired";
169 case VALIDATION_ERR_FUTURE: return "invalid: future";
170 case VALIDATION_ERR_DUPLICATE: return "duplicate";
171 case VALIDATION_ERR_POW: return "pow: insufficient";
172 case VALIDATION_ERR_BLOCKED: return "blocked";
173 case VALIDATION_ERR_TOO_OLD: return "invalid: too old";
174 default: return "error: unknown";
175 }
176}
diff --git a/components/wisp_relay/relay_validator.h b/components/wisp_relay/relay_validator.h
new file mode 100644
index 0000000..c07308f
--- /dev/null
+++ b/components/wisp_relay/relay_validator.h
@@ -0,0 +1,45 @@
1#ifndef RELAY_VALIDATOR_H
2#define RELAY_VALIDATOR_H
3
4#include <stdint.h>
5#include <stdbool.h>
6#include <stddef.h>
7
8typedef enum {
9 VALIDATION_OK = 0,
10 VALIDATION_ERR_SCHEMA,
11 VALIDATION_ERR_ID,
12 VALIDATION_ERR_SIG,
13 VALIDATION_ERR_EXPIRED,
14 VALIDATION_ERR_FUTURE,
15 VALIDATION_ERR_DUPLICATE,
16 VALIDATION_ERR_POW,
17 VALIDATION_ERR_BLOCKED,
18 VALIDATION_ERR_TOO_OLD,
19} validation_result_t;
20
21typedef struct {
22 uint32_t max_event_age_sec;
23 int64_t max_future_sec;
24 uint8_t min_pow_difficulty;
25 bool check_duplicates;
26} validator_config_t;
27
28typedef struct relay_event relay_event_t;
29typedef struct storage_engine storage_engine_t;
30
31validation_result_t relay_validator_check(const uint8_t *id,
32 const uint8_t *pubkey,
33 uint64_t created_at,
34 int kind,
35 const char *content,
36 size_t content_len,
37 const char *tags_json,
38 const uint8_t *sig,
39 const validator_config_t *config);
40
41bool relay_validator_verify_event(const char *event_json, size_t event_len);
42
43const char *relay_validator_result_string(validation_result_t result);
44
45#endif
diff --git a/components/wisp_relay/router.c b/components/wisp_relay/router.c
new file mode 100644
index 0000000..05aa7d4
--- /dev/null
+++ b/components/wisp_relay/router.c
@@ -0,0 +1,140 @@
1#include "router.h"
2#include "ws_server.h"
3#include "handlers.h"
4#include "sub_manager.h"
5#include "cJSON.h"
6#include "esp_log.h"
7#include <string.h>
8
9static const char *TAG = "router";
10
11esp_err_t router_send_notice(relay_ctx_t *ctx, int conn_fd, const char *message)
12{
13 cJSON *arr = cJSON_CreateArray();
14 cJSON_AddItemToArray(arr, cJSON_CreateString("NOTICE"));
15 cJSON_AddItemToArray(arr, cJSON_CreateString(message));
16 char *json = cJSON_PrintUnformatted(arr);
17 cJSON_Delete(arr);
18 esp_err_t ret = ws_server_send(&ctx->ws_server, conn_fd, json, strlen(json));
19 cJSON_free(json);
20 return ret;
21}
22
23esp_err_t router_send_ok(relay_ctx_t *ctx, int conn_fd, const char *event_id_hex,
24 bool accepted, const char *message)
25{
26 cJSON *arr = cJSON_CreateArray();
27 cJSON_AddItemToArray(arr, cJSON_CreateString("OK"));
28 cJSON_AddItemToArray(arr, cJSON_CreateString(event_id_hex));
29 cJSON_AddItemToArray(arr, cJSON_CreateBool(accepted));
30 cJSON_AddItemToArray(arr, cJSON_CreateString(message ? message : ""));
31 char *json = cJSON_PrintUnformatted(arr);
32 cJSON_Delete(arr);
33 esp_err_t ret = ws_server_send(&ctx->ws_server, conn_fd, json, strlen(json));
34 cJSON_free(json);
35 return ret;
36}
37
38esp_err_t router_send_eose(relay_ctx_t *ctx, int conn_fd, const char *sub_id)
39{
40 cJSON *arr = cJSON_CreateArray();
41 cJSON_AddItemToArray(arr, cJSON_CreateString("EOSE"));
42 cJSON_AddItemToArray(arr, cJSON_CreateString(sub_id));
43 char *json = cJSON_PrintUnformatted(arr);
44 cJSON_Delete(arr);
45 esp_err_t ret = ws_server_send(&ctx->ws_server, conn_fd, json, strlen(json));
46 cJSON_free(json);
47 return ret;
48}
49
50esp_err_t router_send_closed(relay_ctx_t *ctx, int conn_fd, const char *sub_id,
51 const char *message)
52{
53 cJSON *arr = cJSON_CreateArray();
54 cJSON_AddItemToArray(arr, cJSON_CreateString("CLOSED"));
55 cJSON_AddItemToArray(arr, cJSON_CreateString(sub_id));
56 cJSON_AddItemToArray(arr, cJSON_CreateString(message ? message : ""));
57 char *json = cJSON_PrintUnformatted(arr);
58 cJSON_Delete(arr);
59 esp_err_t ret = ws_server_send(&ctx->ws_server, conn_fd, json, strlen(json));
60 cJSON_free(json);
61 return ret;
62}
63
64esp_err_t router_send_event(relay_ctx_t *ctx, int conn_fd, const char *sub_id,
65 const char *event_json, size_t event_len)
66{
67 size_t buf_size = event_len + strlen(sub_id) + 32;
68 char *buf = malloc(buf_size);
69 if (!buf) return ESP_ERR_NO_MEM;
70 int n = snprintf(buf, buf_size, "[\"EVENT\",\"%s\",%.*s]", sub_id, (int)event_len, event_json);
71 esp_err_t ret = ws_server_send(&ctx->ws_server, conn_fd, buf, n);
72 free(buf);
73 return ret;
74}
75
76static void on_ws_message(int fd, const char *data, size_t len)
77{
78 extern relay_ctx_t g_relay_ctx;
79 router_dispatch(&g_relay_ctx, fd, data, len);
80}
81
82static void on_ws_disconnect(int fd)
83{
84 extern relay_ctx_t g_relay_ctx;
85 if (g_relay_ctx.sub_manager) {
86 sub_manager_remove_all(g_relay_ctx.sub_manager, fd);
87 }
88}
89
90void router_dispatch(relay_ctx_t *ctx, int conn_fd, const char *data, size_t len)
91{
92 cJSON *arr = cJSON_ParseWithLength(data, len);
93 if (!arr || !cJSON_IsArray(arr)) {
94 router_send_notice(ctx, conn_fd, "invalid JSON");
95 if (arr) cJSON_Delete(arr);
96 return;
97 }
98
99 int array_size = cJSON_GetArraySize(arr);
100 if (array_size < 2) {
101 router_send_notice(ctx, conn_fd, "array too short");
102 cJSON_Delete(arr);
103 return;
104 }
105
106 cJSON *cmd = cJSON_GetArrayItem(arr, 0);
107 if (!cmd || !cJSON_IsString(cmd)) {
108 router_send_notice(ctx, conn_fd, "invalid command");
109 cJSON_Delete(arr);
110 return;
111 }
112
113 const char *cmd_str = cmd->valuestring;
114
115 if (strcmp(cmd_str, "EVENT") == 0 && array_size >= 2) {
116 cJSON *event_obj = cJSON_GetArrayItem(arr, 1);
117 if (event_obj) {
118 char *event_json = cJSON_PrintUnformatted(event_obj);
119 handle_event(ctx, conn_fd, event_json, strlen(event_json));
120 cJSON_free(event_json);
121 }
122 } else if (strcmp(cmd_str, "REQ") == 0 && array_size >= 3) {
123 cJSON *sub_id_item = cJSON_GetArrayItem(arr, 1);
124 if (sub_id_item && cJSON_IsString(sub_id_item)) {
125 cJSON *filter_obj = cJSON_GetArrayItem(arr, 2);
126 char *filter_json = filter_obj ? cJSON_PrintUnformatted(filter_obj) : strdup("{}");
127 handle_req(ctx, conn_fd, sub_id_item->valuestring, filter_json);
128 free(filter_json);
129 }
130 } else if (strcmp(cmd_str, "CLOSE") == 0 && array_size >= 2) {
131 cJSON *sub_id_item = cJSON_GetArrayItem(arr, 1);
132 if (sub_id_item && cJSON_IsString(sub_id_item)) {
133 handle_close(ctx, conn_fd, sub_id_item->valuestring);
134 }
135 } else {
136 router_send_notice(ctx, conn_fd, "unknown command");
137 }
138
139 cJSON_Delete(arr);
140}
diff --git a/components/wisp_relay/router.h b/components/wisp_relay/router.h
new file mode 100644
index 0000000..9afd46e
--- /dev/null
+++ b/components/wisp_relay/router.h
@@ -0,0 +1,19 @@
1#ifndef ROUTER_H
2#define ROUTER_H
3
4#include "relay_core.h"
5#include <stdint.h>
6#include <stddef.h>
7
8esp_err_t router_send_notice(relay_ctx_t *ctx, int conn_fd, const char *message);
9esp_err_t router_send_ok(relay_ctx_t *ctx, int conn_fd, const char *event_id_hex,
10 bool accepted, const char *message);
11esp_err_t router_send_eose(relay_ctx_t *ctx, int conn_fd, const char *sub_id);
12esp_err_t router_send_closed(relay_ctx_t *ctx, int conn_fd, const char *sub_id,
13 const char *message);
14esp_err_t router_send_event(relay_ctx_t *ctx, int conn_fd, const char *sub_id,
15 const char *event_json, size_t event_len);
16
17void router_dispatch(relay_ctx_t *ctx, int conn_fd, const char *data, size_t len);
18
19#endif
diff --git a/components/wisp_relay/storage_engine.c b/components/wisp_relay/storage_engine.c
new file mode 100644
index 0000000..d26705b
--- /dev/null
+++ b/components/wisp_relay/storage_engine.c
@@ -0,0 +1,402 @@
1#include "storage_engine.h"
2#include "esp_littlefs.h"
3#include "esp_log.h"
4#include "nvs_flash.h"
5#include "nvs.h"
6#include <inttypes.h>
7#include <string.h>
8#include <stdio.h>
9#include <sys/stat.h>
10#include <time.h>
11#include <unistd.h>
12
13static const char *TAG = "storage";
14
15#define INDEX_NVS_NAMESPACE "nostr_idx"
16#define EVENTS_DIR "/littlefs/events"
17
18static void get_event_path(const uint8_t event_id[32], uint32_t file_index,
19 char *path, size_t len)
20{
21 char id_hex[33];
22 for (int i = 0; i < 16; i++) sprintf(id_hex + i * 2, "%02x", event_id[i]);
23 snprintf(path, len, EVENTS_DIR "/%02x/%s_%08" PRIx32 ".json",
24 event_id[0], id_hex, file_index);
25}
26
27static int save_index_to_nvs(storage_engine_t *engine)
28{
29 nvs_handle_t nvs;
30 esp_err_t err = nvs_open(INDEX_NVS_NAMESPACE, NVS_READWRITE, &nvs);
31 if (err != ESP_OK) return STORAGE_ERR_IO;
32
33 nvs_set_u16(nvs, "count", engine->index_count);
34 nvs_set_u32(nvs, "next_idx", engine->next_file_index);
35
36 const uint16_t chunk_size = 50;
37 for (uint16_t i = 0; i < engine->index_count; i += chunk_size) {
38 char key[16];
39 snprintf(key, sizeof(key), "idx_%u", i / chunk_size);
40 uint16_t entries = engine->index_count - i;
41 if (entries > chunk_size) entries = chunk_size;
42 nvs_set_blob(nvs, key, &engine->index[i], entries * sizeof(storage_index_entry_t));
43 }
44 nvs_commit(nvs);
45 nvs_close(nvs);
46 return STORAGE_OK;
47}
48
49static int load_index_from_nvs(storage_engine_t *engine)
50{
51 nvs_handle_t nvs;
52 esp_err_t err = nvs_open(INDEX_NVS_NAMESPACE, NVS_READONLY, &nvs);
53 if (err == ESP_ERR_NVS_NOT_FOUND) return STORAGE_OK;
54 if (err != ESP_OK) return STORAGE_ERR_IO;
55
56 err = nvs_get_u16(nvs, "count", &engine->index_count);
57 if (err != ESP_OK) { nvs_close(nvs); return STORAGE_ERR_IO; }
58 if (engine->index_count > engine->max_index_entries) engine->index_count = engine->max_index_entries;
59
60 err = nvs_get_u32(nvs, "next_idx", &engine->next_file_index);
61 if (err != ESP_OK) { nvs_close(nvs); return STORAGE_ERR_IO; }
62
63 const uint16_t chunk_size = 50;
64 for (uint16_t i = 0; i < engine->index_count; i += chunk_size) {
65 char key[16];
66 snprintf(key, sizeof(key), "idx_%u", i / chunk_size);
67 uint16_t entries = engine->index_count - i;
68 if (entries > chunk_size) entries = chunk_size;
69 size_t len = entries * sizeof(storage_index_entry_t);
70 nvs_get_blob(nvs, key, &engine->index[i], &len);
71 }
72 nvs_close(nvs);
73 return STORAGE_OK;
74}
75
76static storage_index_entry_t *find_index_entry(storage_engine_t *engine,
77 const uint8_t event_id[32])
78{
79 for (uint16_t i = 0; i < engine->index_count; i++) {
80 if (memcmp(engine->index[i].event_id, event_id, 32) == 0 &&
81 !(engine->index[i].flags & STORAGE_FLAG_DELETED)) {
82 return &engine->index[i];
83 }
84 }
85 return NULL;
86}
87
88static void parse_event_meta(const char *json, size_t len,
89 uint8_t *id_out, uint8_t *pubkey_out,
90 uint64_t *created_at_out, int *kind_out)
91{
92 extern int relay_hex_to_bytes(const char *hex, size_t hex_len, uint8_t *out, size_t out_len);
93 extern void relay_bytes_to_hex(const uint8_t *bytes, size_t len, char *hex);
94
95 id_out[0] = 0; pubkey_out[0] = 0; *created_at_out = 0; *kind_out = 0;
96
97 const char *p;
98 p = strstr(json, "\"id\":\"");
99 if (p) relay_hex_to_bytes(p + 6, 64, id_out, 32);
100 p = strstr(json, "\"pubkey\":\"");
101 if (p) relay_hex_to_bytes(p + 10, 64, pubkey_out, 32);
102 p = strstr(json, "\"created_at\":");
103 if (p) *created_at_out = strtoull(p + 13, NULL, 10);
104 p = strstr(json, "\"kind\":");
105 if (p) *kind_out = atoi(p + 7);
106}
107
108esp_err_t storage_init(storage_engine_t *engine, uint32_t default_ttl_sec)
109{
110 memset(engine, 0, sizeof(storage_engine_t));
111 engine->default_ttl_sec = default_ttl_sec;
112 strcpy(engine->mount_point, "/littlefs");
113
114 engine->lock = xSemaphoreCreateMutex();
115 if (!engine->lock) return ESP_ERR_NO_MEM;
116
117 engine->max_index_entries = STORAGE_INDEX_ENTRIES;
118 engine->index = heap_caps_calloc(engine->max_index_entries,
119 sizeof(storage_index_entry_t),
120 MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT);
121 if (!engine->index) {
122 engine->max_index_entries = 1000;
123 engine->index = calloc(engine->max_index_entries, sizeof(storage_index_entry_t));
124 if (!engine->index) { vSemaphoreDelete(engine->lock); return ESP_ERR_NO_MEM; }
125 }
126
127 esp_vfs_littlefs_conf_t conf = {
128 .base_path = "/littlefs",
129 .partition_label = STORAGE_PARTITION_LABEL,
130 .format_if_mount_failed = true,
131 .dont_mount = false,
132 };
133
134 esp_err_t ret = esp_vfs_littlefs_register(&conf);
135 if (ret != ESP_OK) {
136 ESP_LOGE(TAG, "Failed to mount LittleFS: %s", esp_err_to_name(ret));
137 free(engine->index);
138 vSemaphoreDelete(engine->lock);
139 return ret;
140 }
141
142 mkdir(EVENTS_DIR, 0755);
143 for (int i = 0; i < 256; i++) {
144 char subdir[64];
145 snprintf(subdir, sizeof(subdir), EVENTS_DIR "/%02x", i);
146 mkdir(subdir, 0755);
147 }
148
149 int load_err = load_index_from_nvs(engine);
150 if (load_err != STORAGE_OK) {
151 ESP_LOGW(TAG, "Failed to load index, starting fresh");
152 engine->index_count = 0;
153 engine->next_file_index = 0;
154 }
155
156 engine->initialized = true;
157
158 size_t total, used;
159 esp_littlefs_info(STORAGE_PARTITION_LABEL, &total, &used);
160 ESP_LOGI(TAG, "Storage initialized: %" PRIu16 " events, %zu/%zu bytes used",
161 engine->index_count, used, total);
162 return ESP_OK;
163}
164
165void storage_destroy(storage_engine_t *engine)
166{
167 if (!engine->initialized) return;
168 if (engine->cleanup_task) {
169 engine->cleanup_stop = true;
170 while (engine->cleanup_task != NULL) vTaskDelay(pdMS_TO_TICKS(100));
171 }
172 save_index_to_nvs(engine);
173 esp_vfs_littlefs_unregister(STORAGE_PARTITION_LABEL);
174 if (engine->index) { free(engine->index); engine->index = NULL; }
175 if (engine->lock) { vSemaphoreDelete(engine->lock); engine->lock = NULL; }
176 engine->initialized = false;
177}
178
179storage_error_t storage_save_event_json(storage_engine_t *engine,
180 const char *event_json,
181 size_t event_json_len)
182{
183 if (!engine->initialized) return STORAGE_ERR_NOT_INITIALIZED;
184
185 uint8_t id[32] = {0}, pubkey[32] = {0};
186 uint64_t created_at = 0;
187 int kind = 0;
188 parse_event_meta(event_json, event_json_len, id, pubkey, &created_at, &kind);
189
190 xSemaphoreTake(engine->lock, portMAX_DELAY);
191
192 if (find_index_entry(engine, id)) {
193 xSemaphoreGive(engine->lock);
194 return STORAGE_ERR_DUPLICATE;
195 }
196 if (engine->index_count >= engine->max_index_entries) {
197 xSemaphoreGive(engine->lock);
198 return STORAGE_ERR_FULL;
199 }
200
201 char path[128];
202 get_event_path(id, engine->next_file_index, path, sizeof(path));
203 FILE *f = fopen(path, "wb");
204 if (!f) {
205 char dir[64];
206 snprintf(dir, sizeof(dir), EVENTS_DIR "/%02x", id[0]);
207 mkdir(dir, 0755);
208 f = fopen(path, "wb");
209 }
210 if (!f) { xSemaphoreGive(engine->lock); return STORAGE_ERR_IO; }
211
212 fwrite(event_json, 1, event_json_len, f);
213 fclose(f);
214
215 storage_index_entry_t *entry = &engine->index[engine->index_count];
216 memcpy(entry->event_id, id, 32);
217 entry->created_at = (uint32_t)created_at;
218 entry->kind = kind;
219 memcpy(entry->pubkey_prefix, pubkey, 4);
220 entry->file_index = engine->next_file_index;
221 entry->flags = 0;
222 entry->expires_at = (uint32_t)time(NULL) + engine->default_ttl_sec;
223
224 engine->index_count++;
225 engine->next_file_index++;
226 if (engine->index_count % 10 == 0) save_index_to_nvs(engine);
227
228 xSemaphoreGive(engine->lock);
229 return STORAGE_OK;
230}
231
232bool storage_event_exists(storage_engine_t *engine, const uint8_t event_id[32])
233{
234 if (!engine->initialized) return false;
235 xSemaphoreTake(engine->lock, portMAX_DELAY);
236 bool exists = (find_index_entry(engine, event_id) != NULL);
237 xSemaphoreGive(engine->lock);
238 return exists;
239}
240
241storage_error_t storage_query_events_json(storage_engine_t *engine,
242 int kind,
243 const char *author_hex,
244 int limit,
245 char ***results,
246 uint16_t *count)
247{
248 if (!engine->initialized) return STORAGE_ERR_NOT_INITIALIZED;
249 *results = NULL;
250 *count = 0;
251 if (limit > 500) limit = 500;
252 if (limit <= 0) limit = 100;
253
254 char **out = calloc(limit, sizeof(char *));
255 if (!out) return STORAGE_ERR_NO_MEM;
256
257 xSemaphoreTake(engine->lock, portMAX_DELAY);
258 uint32_t now = (uint32_t)time(NULL);
259 uint16_t found = 0;
260
261 uint8_t author_prefix[4] = {0};
262 int have_author = 0;
263 if (author_hex && strlen(author_hex) >= 8) {
264 extern int relay_hex_to_bytes(const char *, size_t, uint8_t *, size_t);
265 relay_hex_to_bytes(author_hex, 8, author_prefix, 4);
266 have_author = 1;
267 }
268
269 for (int i = engine->index_count - 1; i >= 0 && found < limit; i--) {
270 storage_index_entry_t *e = &engine->index[i];
271 if (e->flags & STORAGE_FLAG_DELETED) continue;
272 if (e->expires_at > 0 && e->expires_at < now) continue;
273 if (kind > 0 && e->kind != kind) continue;
274 if (have_author && memcmp(e->pubkey_prefix, author_prefix, 4) != 0) continue;
275
276 char path[128];
277 get_event_path(e->event_id, e->file_index, path, sizeof(path));
278 FILE *f = fopen(path, "rb");
279 if (!f) continue;
280 fseek(f, 0, SEEK_END);
281 long sz = ftell(f);
282 fseek(f, 0, SEEK_SET);
283 if (sz <= 0 || sz > STORAGE_MAX_EVENT_SIZE) { fclose(f); continue; }
284 char *buf = malloc(sz + 1);
285 fread(buf, 1, sz, f);
286 buf[sz] = '\0';
287 fclose(f);
288 out[found++] = buf;
289 }
290
291 xSemaphoreGive(engine->lock);
292 *results = out;
293 *count = found;
294 return STORAGE_OK;
295}
296
297void storage_free_query_results(char **results, uint16_t count)
298{
299 if (!results) return;
300 for (uint16_t i = 0; i < count; i++) free(results[i]);
301 free(results);
302}
303
304storage_error_t storage_delete_event(storage_engine_t *engine, const uint8_t event_id[32])
305{
306 if (!engine->initialized) return STORAGE_ERR_NOT_INITIALIZED;
307 xSemaphoreTake(engine->lock, portMAX_DELAY);
308 storage_index_entry_t *e = find_index_entry(engine, event_id);
309 if (!e) { xSemaphoreGive(engine->lock); return STORAGE_ERR_NOT_FOUND; }
310 char path[128];
311 get_event_path(e->event_id, e->file_index, path, sizeof(path));
312 unlink(path);
313 e->flags |= STORAGE_FLAG_DELETED;
314 save_index_to_nvs(engine);
315 xSemaphoreGive(engine->lock);
316 return STORAGE_OK;
317}
318
319int storage_purge_expired(storage_engine_t *engine)
320{
321 if (!engine->initialized) return 0;
322 xSemaphoreTake(engine->lock, portMAX_DELAY);
323 uint32_t now = (uint32_t)time(NULL);
324 int purged = 0;
325 for (uint16_t i = 0; i < engine->index_count; i++) {
326 if (engine->index[i].flags & STORAGE_FLAG_DELETED) continue;
327 if (engine->index[i].expires_at > 0 && engine->index[i].expires_at < now) {
328 char path[128];
329 get_event_path(engine->index[i].event_id, engine->index[i].file_index, path, sizeof(path));
330 unlink(path);
331 engine->index[i].flags |= STORAGE_FLAG_DELETED;
332 purged++;
333 }
334 }
335 if (purged > 0) { save_index_to_nvs(engine); ESP_LOGI(TAG, "Purged %d expired events", purged); }
336 xSemaphoreGive(engine->lock);
337 return purged;
338}
339
340int storage_compact_index(storage_engine_t *engine)
341{
342 if (!engine->initialized) return 0;
343 xSemaphoreTake(engine->lock, portMAX_DELAY);
344 uint16_t write_idx = 0;
345 int compacted = 0;
346 for (uint16_t read_idx = 0; read_idx < engine->index_count; read_idx++) {
347 if (!(engine->index[read_idx].flags & STORAGE_FLAG_DELETED)) {
348 if (write_idx != read_idx)
349 memcpy(&engine->index[write_idx], &engine->index[read_idx], sizeof(storage_index_entry_t));
350 write_idx++;
351 } else {
352 compacted++;
353 }
354 }
355 if (compacted > 0) {
356 engine->index_count = write_idx;
357 save_index_to_nvs(engine);
358 ESP_LOGI(TAG, "Compacted: removed %d, %" PRIu16 " remaining", compacted, engine->index_count);
359 }
360 xSemaphoreGive(engine->lock);
361 return compacted;
362}
363
364void storage_get_stats(storage_engine_t *engine, storage_stats_t *stats)
365{
366 memset(stats, 0, sizeof(storage_stats_t));
367 if (!engine->initialized) return;
368 xSemaphoreTake(engine->lock, portMAX_DELAY);
369 uint32_t now = (uint32_t)time(NULL);
370 for (uint16_t i = 0; i < engine->index_count; i++) {
371 if (engine->index[i].flags & STORAGE_FLAG_DELETED) continue;
372 if (engine->index[i].expires_at > 0 && engine->index[i].expires_at < now) continue;
373 stats->total_events++;
374 }
375 size_t total, used;
376 esp_littlefs_info(STORAGE_PARTITION_LABEL, &total, &used);
377 stats->total_bytes = total;
378 stats->free_bytes = total - used;
379 xSemaphoreGive(engine->lock);
380}
381
382static void storage_cleanup_task(void *arg)
383{
384 storage_engine_t *engine = (storage_engine_t *)arg;
385 int cycles = 0;
386 while (!engine->cleanup_stop) {
387 for (int i = 0; i < 60 && !engine->cleanup_stop; i++) vTaskDelay(pdMS_TO_TICKS(1000));
388 if (engine->cleanup_stop) break;
389 storage_purge_expired(engine);
390 if (++cycles >= 10) { storage_compact_index(engine); cycles = 0; }
391 }
392 engine->cleanup_task = NULL;
393 vTaskDelete(NULL);
394}
395
396esp_err_t storage_start_cleanup_task(storage_engine_t *engine)
397{
398 engine->cleanup_stop = false;
399 BaseType_t ret = xTaskCreate(storage_cleanup_task, "relay_cleanup", 4096, engine, 2, &engine->cleanup_task);
400 if (ret != pdPASS) { engine->cleanup_task = NULL; return ESP_ERR_NO_MEM; }
401 return ESP_OK;
402}
diff --git a/components/wisp_relay/storage_engine.h b/components/wisp_relay/storage_engine.h
new file mode 100644
index 0000000..4e17113
--- /dev/null
+++ b/components/wisp_relay/storage_engine.h
@@ -0,0 +1,88 @@
1#ifndef STORAGE_ENGINE_H
2#define STORAGE_ENGINE_H
3
4#include <stdbool.h>
5#include <stdint.h>
6#include "esp_err.h"
7#include "freertos/FreeRTOS.h"
8#include "freertos/semphr.h"
9#include "freertos/task.h"
10
11#define STORAGE_MAX_EVENTS 5000
12#define STORAGE_MAX_EVENT_SIZE 8192
13#define STORAGE_INDEX_ENTRIES 5000
14#define STORAGE_PARTITION_LABEL "relay_store"
15
16typedef enum {
17 STORAGE_OK = 0,
18 STORAGE_ERR_NOT_INITIALIZED,
19 STORAGE_ERR_FULL,
20 STORAGE_ERR_DUPLICATE,
21 STORAGE_ERR_NOT_FOUND,
22 STORAGE_ERR_IO,
23 STORAGE_ERR_NO_MEM,
24 STORAGE_ERR_SERIALIZE
25} storage_error_t;
26
27#define STORAGE_FLAG_DELETED 0x01
28
29typedef struct __attribute__((packed)) {
30 uint8_t event_id[32];
31 uint32_t created_at;
32 uint32_t expires_at;
33 uint32_t file_index;
34 uint16_t kind;
35 uint8_t pubkey_prefix[4];
36 uint8_t flags;
37 uint8_t reserved;
38} storage_index_entry_t;
39
40typedef struct {
41 uint32_t total_events;
42 uint32_t total_bytes;
43 uint32_t free_bytes;
44 uint32_t oldest_event_ts;
45 uint32_t newest_event_ts;
46} storage_stats_t;
47
48typedef struct storage_engine {
49 storage_index_entry_t *index;
50 uint16_t index_count;
51 uint16_t max_index_entries;
52 uint32_t next_file_index;
53 SemaphoreHandle_t lock;
54 TaskHandle_t cleanup_task;
55 bool initialized;
56 bool cleanup_stop;
57 char mount_point[16];
58 uint32_t default_ttl_sec;
59} storage_engine_t;
60
61esp_err_t storage_init(storage_engine_t *engine, uint32_t default_ttl_sec);
62void storage_destroy(storage_engine_t *engine);
63
64storage_error_t storage_save_event_json(storage_engine_t *engine,
65 const char *event_json,
66 size_t event_json_len);
67
68storage_error_t storage_query_events_json(storage_engine_t *engine,
69 int kind,
70 const char *author_hex,
71 int limit,
72 char ***results,
73 uint16_t *count);
74
75void storage_free_query_results(char **results, uint16_t count);
76
77bool storage_event_exists(storage_engine_t *engine, const uint8_t event_id[32]);
78
79storage_error_t storage_delete_event(storage_engine_t *engine, const uint8_t event_id[32]);
80
81int storage_purge_expired(storage_engine_t *engine);
82int storage_compact_index(storage_engine_t *engine);
83
84void storage_get_stats(storage_engine_t *engine, storage_stats_t *stats);
85
86esp_err_t storage_start_cleanup_task(storage_engine_t *engine);
87
88#endif
diff --git a/components/wisp_relay/sub_manager.c b/components/wisp_relay/sub_manager.c
new file mode 100644
index 0000000..a1da2e3
--- /dev/null
+++ b/components/wisp_relay/sub_manager.c
@@ -0,0 +1,272 @@
1#include "sub_manager.h"
2#include "relay_types.h"
3#include "esp_log.h"
4#include <string.h>
5#include <stdlib.h>
6
7static const char *TAG = "sub_mgr";
8
9static void filter_clear(sub_filter_t *f)
10{
11 for (size_t i = 0; i < f->ids_count; i++) free(f->ids[i]);
12 for (size_t i = 0; i < f->authors_count; i++) free(f->authors[i]);
13 for (size_t i = 0; i < f->e_tags_count; i++) free(f->e_tags[i]);
14 for (size_t i = 0; i < f->p_tags_count; i++) free(f->p_tags[i]);
15 memset(f, 0, sizeof(sub_filter_t));
16}
17
18static bool filter_copy(sub_filter_t *dst, const sub_filter_t *src)
19{
20 memset(dst, 0, sizeof(sub_filter_t));
21
22 size_t ids_count = src->ids_count > SUB_MAX_FILTER_IDS ? SUB_MAX_FILTER_IDS : src->ids_count;
23 for (size_t i = 0; i < ids_count; i++) {
24 dst->ids[i] = strdup(src->ids[i]);
25 if (!dst->ids[i]) goto fail;
26 }
27 dst->ids_count = ids_count;
28
29 size_t authors_count = src->authors_count > SUB_MAX_FILTER_AUTHORS ? SUB_MAX_FILTER_AUTHORS : src->authors_count;
30 for (size_t i = 0; i < authors_count; i++) {
31 dst->authors[i] = strdup(src->authors[i]);
32 if (!dst->authors[i]) goto fail;
33 }
34 dst->authors_count = authors_count;
35
36 size_t kinds_count = src->kinds_count > SUB_MAX_FILTER_KINDS ? SUB_MAX_FILTER_KINDS : src->kinds_count;
37 memcpy(dst->kinds, src->kinds, kinds_count * sizeof(int32_t));
38 dst->kinds_count = kinds_count;
39
40 size_t e_tags_count = src->e_tags_count > SUB_MAX_FILTER_ETAGS ? SUB_MAX_FILTER_ETAGS : src->e_tags_count;
41 for (size_t i = 0; i < e_tags_count; i++) {
42 dst->e_tags[i] = strdup(src->e_tags[i]);
43 if (!dst->e_tags[i]) goto fail;
44 }
45 dst->e_tags_count = e_tags_count;
46
47 size_t p_tags_count = src->p_tags_count > SUB_MAX_FILTER_PTAGS ? SUB_MAX_FILTER_PTAGS : src->p_tags_count;
48 for (size_t i = 0; i < p_tags_count; i++) {
49 dst->p_tags[i] = strdup(src->p_tags[i]);
50 if (!dst->p_tags[i]) goto fail;
51 }
52 dst->p_tags_count = p_tags_count;
53
54 dst->since = src->since;
55 dst->until = src->until;
56 dst->limit = src->limit;
57 return true;
58
59fail:
60 filter_clear(dst);
61 return false;
62}
63
64static void clear_subscription(subscription_t *sub)
65{
66 for (uint8_t i = 0; i < sub->filter_count; i++) {
67 filter_clear(&sub->filters[i]);
68 }
69 memset(sub, 0, sizeof(subscription_t));
70}
71
72esp_err_t sub_manager_init(sub_manager_t *mgr)
73{
74 memset(mgr, 0, sizeof(sub_manager_t));
75 mgr->lock = xSemaphoreCreateMutex();
76 if (!mgr->lock) return ESP_ERR_NO_MEM;
77 ESP_LOGI(TAG, "Initialized (max=%d, per_conn=%d)", SUB_MAX_TOTAL, SUB_MAX_PER_CONN);
78 return ESP_OK;
79}
80
81void sub_manager_destroy(sub_manager_t *mgr)
82{
83 if (!mgr) return;
84 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
85 if (mgr->subs[i].active) clear_subscription(&mgr->subs[i]);
86 }
87 if (mgr->lock) { vSemaphoreDelete(mgr->lock); mgr->lock = NULL; }
88}
89
90static subscription_t *find_sub(sub_manager_t *mgr, int conn_fd, const char *sub_id)
91{
92 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
93 if (mgr->subs[i].active && mgr->subs[i].conn_fd == conn_fd &&
94 strcmp(mgr->subs[i].sub_id, sub_id) == 0)
95 return &mgr->subs[i];
96 }
97 return NULL;
98}
99
100static subscription_t *find_free_slot(sub_manager_t *mgr)
101{
102 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
103 if (!mgr->subs[i].active) return &mgr->subs[i];
104 }
105 return NULL;
106}
107
108static bool hex_prefix_match(const char *prefix, size_t prefix_len,
109 const char *full, size_t full_len)
110{
111 if (prefix_len == 0) return true;
112 if (prefix_len > full_len) return false;
113 return memcmp(prefix, full, prefix_len) == 0;
114}
115
116static bool filter_matches_event(const sub_filter_t *f, int event_kind,
117 const char *pubkey_hex, uint64_t created_at)
118{
119 if (f->kinds_count > 0) {
120 bool found = false;
121 for (size_t i = 0; i < f->kinds_count; i++) {
122 if (f->kinds[i] == event_kind) { found = true; break; }
123 }
124 if (!found) return false;
125 }
126
127 if (f->authors_count > 0) {
128 bool found = false;
129 for (size_t i = 0; i < f->authors_count; i++) {
130 if (hex_prefix_match(f->authors[i], strlen(f->authors[i]),
131 pubkey_hex, strlen(pubkey_hex))) {
132 found = true; break;
133 }
134 }
135 if (!found) return false;
136 }
137
138 if (f->since > 0 && (int64_t)created_at < f->since) return false;
139 if (f->until > 0 && (int64_t)created_at > f->until) return false;
140
141 return true;
142}
143
144void sub_manager_match_json(sub_manager_t *mgr, const char *event_json,
145 size_t event_len, int event_kind,
146 const char *event_pubkey_hex,
147 uint64_t event_created_at,
148 sub_match_result_t *result)
149{
150 result->count = 0;
151 (void)event_json;
152 (void)event_len;
153
154 xSemaphoreTake(mgr->lock, portMAX_DELAY);
155 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
156 subscription_t *sub = &mgr->subs[i];
157 if (!sub->active) continue;
158
159 bool matched = false;
160 for (uint8_t f = 0; f < sub->filter_count; f++) {
161 if (filter_matches_event(&sub->filters[f], event_kind,
162 event_pubkey_hex, event_created_at)) {
163 matched = true;
164 break;
165 }
166 }
167 if (matched) {
168 sub_match_entry_t *entry = &result->matches[result->count++];
169 entry->conn_fd = sub->conn_fd;
170 memcpy(entry->sub_id, sub->sub_id, sizeof(entry->sub_id));
171 }
172 }
173 xSemaphoreGive(mgr->lock);
174}
175
176sub_error_t sub_manager_add(sub_manager_t *mgr, int conn_fd,
177 const char *sub_id,
178 const sub_filter_t *filters,
179 size_t filter_count)
180{
181 if (filter_count > SUB_MAX_FILTERS) filter_count = SUB_MAX_FILTERS;
182
183 xSemaphoreTake(mgr->lock, portMAX_DELAY);
184
185 subscription_t *existing = find_sub(mgr, conn_fd, sub_id);
186 if (existing) {
187 for (uint8_t i = 0; i < existing->filter_count; i++)
188 filter_clear(&existing->filters[i]);
189 existing->events_sent = 0;
190 for (size_t i = 0; i < filter_count; i++) {
191 if (!filter_copy(&existing->filters[i], &filters[i])) {
192 existing->filter_count = (uint8_t)i;
193 xSemaphoreGive(mgr->lock);
194 return SUB_ERR_MEMORY;
195 }
196 }
197 existing->filter_count = (uint8_t)filter_count;
198 xSemaphoreGive(mgr->lock);
199 return SUB_OK;
200 }
201
202 uint8_t conn_count = 0;
203 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
204 if (mgr->subs[i].active && mgr->subs[i].conn_fd == conn_fd) conn_count++;
205 }
206 if (conn_count >= SUB_MAX_PER_CONN) {
207 xSemaphoreGive(mgr->lock);
208 return SUB_ERR_TOO_MANY_FILTERS;
209 }
210
211 subscription_t *slot = find_free_slot(mgr);
212 if (!slot) { xSemaphoreGive(mgr->lock); return SUB_ERR_MEMORY; }
213
214 memset(slot, 0, sizeof(subscription_t));
215 strncpy(slot->sub_id, sub_id, SUB_MAX_ID_LEN);
216 slot->sub_id[SUB_MAX_ID_LEN] = '\0';
217 slot->conn_fd = conn_fd;
218
219 for (size_t i = 0; i < filter_count; i++) {
220 if (!filter_copy(&slot->filters[i], &filters[i])) {
221 slot->filter_count = (uint8_t)i;
222 clear_subscription(slot);
223 xSemaphoreGive(mgr->lock);
224 return SUB_ERR_MEMORY;
225 }
226 }
227 slot->filter_count = (uint8_t)filter_count;
228 slot->active = true;
229 mgr->active_count++;
230
231 ESP_LOGI(TAG, "Added sub=%s fd=%d filters=%zu total=%d",
232 sub_id, conn_fd, filter_count, mgr->active_count);
233 xSemaphoreGive(mgr->lock);
234 return SUB_OK;
235}
236
237sub_error_t sub_manager_remove(sub_manager_t *mgr, int conn_fd, const char *sub_id)
238{
239 xSemaphoreTake(mgr->lock, portMAX_DELAY);
240 subscription_t *sub = find_sub(mgr, conn_fd, sub_id);
241 if (!sub) { xSemaphoreGive(mgr->lock); return SUB_ERR_NOT_FOUND; }
242 clear_subscription(sub);
243 mgr->active_count--;
244 xSemaphoreGive(mgr->lock);
245 return SUB_OK;
246}
247
248void sub_manager_remove_all(sub_manager_t *mgr, int conn_fd)
249{
250 xSemaphoreTake(mgr->lock, portMAX_DELAY);
251 int removed = 0;
252 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
253 if (mgr->subs[i].active && mgr->subs[i].conn_fd == conn_fd) {
254 clear_subscription(&mgr->subs[i]);
255 mgr->active_count--;
256 removed++;
257 }
258 }
259 if (removed > 0) ESP_LOGI(TAG, "Removed %d subs for fd=%d", removed, conn_fd);
260 xSemaphoreGive(mgr->lock);
261}
262
263uint8_t sub_manager_count(sub_manager_t *mgr, int conn_fd)
264{
265 uint8_t count = 0;
266 xSemaphoreTake(mgr->lock, portMAX_DELAY);
267 for (int i = 0; i < SUB_MAX_TOTAL; i++) {
268 if (mgr->subs[i].active && mgr->subs[i].conn_fd == conn_fd) count++;
269 }
270 xSemaphoreGive(mgr->lock);
271 return count;
272}
diff --git a/components/wisp_relay/sub_manager.h b/components/wisp_relay/sub_manager.h
new file mode 100644
index 0000000..64afb04
--- /dev/null
+++ b/components/wisp_relay/sub_manager.h
@@ -0,0 +1,92 @@
1#ifndef SUB_MANAGER_H
2#define SUB_MANAGER_H
3
4#include <stdbool.h>
5#include <stdint.h>
6#include "esp_err.h"
7#include "freertos/FreeRTOS.h"
8#include "freertos/semphr.h"
9#include "relay_types.h"
10
11#define SUB_MAX_TOTAL 64
12#define SUB_MAX_PER_CONN 8
13#define SUB_MAX_FILTERS 4
14#define SUB_MAX_ID_LEN 64
15
16#define SUB_MAX_FILTER_IDS 20
17#define SUB_MAX_FILTER_AUTHORS 20
18#define SUB_MAX_FILTER_KINDS 20
19#define SUB_MAX_FILTER_ETAGS 20
20#define SUB_MAX_FILTER_PTAGS 20
21
22typedef enum {
23 SUB_OK = 0,
24 SUB_ERR_INVALID,
25 SUB_ERR_TOO_MANY_FILTERS,
26 SUB_ERR_MEMORY,
27 SUB_ERR_NOT_FOUND,
28} sub_error_t;
29
30typedef struct {
31 char *ids[SUB_MAX_FILTER_IDS];
32 size_t ids_count;
33 char *authors[SUB_MAX_FILTER_AUTHORS];
34 size_t authors_count;
35 int32_t kinds[SUB_MAX_FILTER_KINDS];
36 size_t kinds_count;
37 char *e_tags[SUB_MAX_FILTER_ETAGS];
38 size_t e_tags_count;
39 char *p_tags[SUB_MAX_FILTER_PTAGS];
40 size_t p_tags_count;
41 int64_t since;
42 int64_t until;
43 int limit;
44} sub_filter_t;
45
46typedef struct {
47 char sub_id[SUB_MAX_ID_LEN + 1];
48 int conn_fd;
49 sub_filter_t filters[SUB_MAX_FILTERS];
50 uint8_t filter_count;
51 uint16_t events_sent;
52 bool active;
53} subscription_t;
54
55typedef struct sub_manager {
56 subscription_t subs[SUB_MAX_TOTAL];
57 SemaphoreHandle_t lock;
58 uint16_t active_count;
59} sub_manager_t;
60
61typedef struct {
62 int conn_fd;
63 char sub_id[SUB_MAX_ID_LEN + 1];
64} sub_match_entry_t;
65
66typedef struct {
67 sub_match_entry_t matches[SUB_MAX_TOTAL];
68 uint8_t count;
69} sub_match_result_t;
70
71esp_err_t sub_manager_init(sub_manager_t *mgr);
72void sub_manager_destroy(sub_manager_t *mgr);
73
74sub_error_t sub_manager_add(sub_manager_t *mgr, int conn_fd,
75 const char *sub_id,
76 const sub_filter_t *filters,
77 size_t filter_count);
78
79sub_error_t sub_manager_remove(sub_manager_t *mgr, int conn_fd,
80 const char *sub_id);
81
82void sub_manager_remove_all(sub_manager_t *mgr, int conn_fd);
83
84void sub_manager_match_json(sub_manager_t *mgr, const char *event_json,
85 size_t event_len, int event_kind,
86 const char *event_pubkey_hex,
87 uint64_t event_created_at,
88 sub_match_result_t *result);
89
90uint8_t sub_manager_count(sub_manager_t *mgr, int conn_fd);
91
92#endif
diff --git a/components/wisp_relay/ws_server.c b/components/wisp_relay/ws_server.c
new file mode 100644
index 0000000..a973ca6
--- /dev/null
+++ b/components/wisp_relay/ws_server.c
@@ -0,0 +1,426 @@
1#include "ws_server.h"
2#include "nip11_relay.h"
3#include "esp_log.h"
4#include "esp_timer.h"
5#include <string.h>
6#include <strings.h>
7#include <unistd.h>
8#include <sys/socket.h>
9#include <netinet/in.h>
10#include <netinet/tcp.h>
11#include <arpa/inet.h>
12
13static const char *TAG = "ws_server";
14static ws_message_cb_t g_message_callback = NULL;
15static ws_disconnect_cb_t g_disconnect_callback = NULL;
16static ws_server_t *g_server = NULL;
17static __thread httpd_req_t *g_current_req = NULL;
18
19static ws_connection_t* find_free_slot(ws_server_t *server)
20{
21 for (int i = 0; i < WS_MAX_CONNECTIONS; i++) {
22 if (!server->connections[i].active) {
23 return &server->connections[i];
24 }
25 }
26 return NULL;
27}
28
29static ws_connection_t* find_connection_by_fd(ws_server_t *server, int fd)
30{
31 for (int i = 0; i < WS_MAX_CONNECTIONS; i++) {
32 if (server->connections[i].active && server->connections[i].fd == fd) {
33 return &server->connections[i];
34 }
35 }
36 return NULL;
37}
38
39static void update_connection_activity(ws_server_t *server, int fd)
40{
41 xSemaphoreTake(server->lock, portMAX_DELAY);
42 ws_connection_t *conn = find_connection_by_fd(server, fd);
43 if (conn) {
44 conn->last_activity = esp_timer_get_time() / 1000000;
45 }
46 xSemaphoreGive(server->lock);
47}
48
49static void set_unknown_ip(char *ip_buf, size_t buf_len)
50{
51 if (buf_len == 0) {
52 return;
53 }
54 strncpy(ip_buf, "unknown", buf_len - 1);
55 ip_buf[buf_len - 1] = '\0';
56}
57
58static void get_client_ip(int fd, char *ip_buf, size_t buf_len)
59{
60 if (buf_len == 0) {
61 return;
62 }
63
64 struct sockaddr_storage addr;
65 socklen_t addr_len = sizeof(addr);
66
67 if (getpeername(fd, (struct sockaddr *)&addr, &addr_len) != 0) {
68 set_unknown_ip(ip_buf, buf_len);
69 return;
70 }
71
72 const char *result = NULL;
73 if (addr.ss_family == AF_INET) {
74 struct sockaddr_in *addr_in = (struct sockaddr_in *)&addr;
75 result = inet_ntop(AF_INET, &addr_in->sin_addr, ip_buf, buf_len);
76 }
77 if (!result) {
78 set_unknown_ip(ip_buf, buf_len);
79 }
80}
81
82static esp_err_t on_open(httpd_handle_t hd, int sockfd)
83{
84 if (!g_server) return ESP_FAIL;
85
86 xSemaphoreTake(g_server->lock, portMAX_DELAY);
87
88 if (g_server->connection_count >= WS_MAX_CONNECTIONS) {
89 xSemaphoreGive(g_server->lock);
90 ESP_LOGW(TAG, "Connection rejected - max connections reached");
91 return ESP_FAIL;
92 }
93
94 ws_connection_t *conn = find_free_slot(g_server);
95 if (!conn) {
96 xSemaphoreGive(g_server->lock);
97 ESP_LOGE(TAG, "No free slot despite connection_count < WS_MAX_CONNECTIONS (fd=%d)", sockfd);
98 return ESP_FAIL;
99 }
100
101 struct linger so_linger = { .l_onoff = 1, .l_linger = 0 };
102 setsockopt(sockfd, SOL_SOCKET, SO_LINGER, &so_linger, sizeof(so_linger));
103
104 int nodelay = 1;
105 setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay));
106
107 conn->fd = sockfd;
108 conn->active = true;
109 conn->connected_at = esp_timer_get_time() / 1000000;
110 conn->last_activity = conn->connected_at;
111 get_client_ip(sockfd, conn->remote_ip, sizeof(conn->remote_ip));
112 g_server->connection_count++;
113 ESP_LOGI(TAG, "New connection from %s (fd=%d, total=%d)",
114 conn->remote_ip, sockfd, g_server->connection_count);
115
116 xSemaphoreGive(g_server->lock);
117 return ESP_OK;
118}
119
120static void on_close(httpd_handle_t hd, int sockfd)
121{
122 if (!g_server) return;
123
124 if (g_disconnect_callback) {
125 g_disconnect_callback(sockfd);
126 }
127
128 xSemaphoreTake(g_server->lock, portMAX_DELAY);
129
130 ws_connection_t *conn = find_connection_by_fd(g_server, sockfd);
131 if (conn) {
132 ESP_LOGI(TAG, "Connection closed (fd=%d, ip=%s)", sockfd, conn->remote_ip);
133 memset(conn, 0, sizeof(ws_connection_t));
134 g_server->connection_count--;
135 }
136
137 xSemaphoreGive(g_server->lock);
138}
139
140void ws_server_set_disconnect_cb(ws_disconnect_cb_t cb)
141{
142 g_disconnect_callback = cb;
143}
144
145static esp_err_t ws_handler(httpd_req_t *req)
146{
147 if (req->method == HTTP_GET) {
148 char upgrade[16] = {0};
149 if (httpd_req_get_hdr_value_str(req, "Upgrade", upgrade, sizeof(upgrade)) != ESP_OK ||
150 strcasecmp(upgrade, "websocket") != 0) {
151 return relay_nip11_handler(req);
152 }
153 ESP_LOGD(TAG, "WebSocket handshake completed");
154 return ESP_OK;
155 }
156
157 httpd_ws_frame_t ws_pkt;
158 memset(&ws_pkt, 0, sizeof(httpd_ws_frame_t));
159 ws_pkt.type = HTTPD_WS_TYPE_TEXT;
160
161 esp_err_t ret = httpd_ws_recv_frame(req, &ws_pkt, 0);
162 if (ret != ESP_OK) {
163 ESP_LOGE(TAG, "Failed to get frame len: %d", ret);
164 return ret;
165 }
166
167 if (ws_pkt.len == 0) {
168 return ESP_OK;
169 }
170
171 if (ws_pkt.len > WS_MAX_FRAME_SIZE) {
172 ESP_LOGW(TAG, "Frame too large: %zu bytes", ws_pkt.len);
173 return ESP_FAIL;
174 }
175
176 ws_pkt.payload = malloc(ws_pkt.len + 1);
177 if (!ws_pkt.payload) {
178 ESP_LOGE(TAG, "Failed to allocate %zu bytes", ws_pkt.len);
179 return ESP_ERR_NO_MEM;
180 }
181
182 ret = httpd_ws_recv_frame(req, &ws_pkt, ws_pkt.len);
183 if (ret != ESP_OK) {
184 ESP_LOGE(TAG, "Failed to receive frame: %d", ret);
185 free(ws_pkt.payload);
186 return ret;
187 }
188
189 ((char *)ws_pkt.payload)[ws_pkt.len] = '\0';
190
191 int fd = httpd_req_to_sockfd(req);
192 if (g_server) {
193 update_connection_activity(g_server, fd);
194 }
195
196 switch (ws_pkt.type) {
197 case HTTPD_WS_TYPE_TEXT:
198 ESP_LOGD(TAG, "Received %zu bytes from fd=%d", ws_pkt.len, fd);
199 if (g_message_callback) {
200 g_current_req = req;
201 g_message_callback(fd, (char *)ws_pkt.payload, ws_pkt.len);
202 g_current_req = NULL;
203 }
204 break;
205
206 case HTTPD_WS_TYPE_PING:
207 ws_pkt.type = HTTPD_WS_TYPE_PONG;
208 ret = httpd_ws_send_frame(req, &ws_pkt);
209 if (ret != ESP_OK) {
210 ESP_LOGW(TAG, "Failed to send PONG to fd=%d: %d", fd, ret);
211 free(ws_pkt.payload);
212 return ret;
213 }
214 break;
215
216 case HTTPD_WS_TYPE_CLOSE: {
217 ESP_LOGD(TAG, "Received CLOSE frame from fd=%d", fd);
218 free(ws_pkt.payload);
219 httpd_ws_frame_t close_pkt = {
220 .type = HTTPD_WS_TYPE_CLOSE,
221 .payload = NULL,
222 .len = 0,
223 };
224 httpd_ws_send_frame(req, &close_pkt);
225 return ESP_FAIL;
226 }
227
228 default:
229 break;
230 }
231
232 free(ws_pkt.payload);
233 return ESP_OK;
234}
235
236typedef struct {
237 httpd_handle_t hd;
238 int fd;
239 char *data;
240 size_t len;
241} async_send_arg_t;
242
243static void ws_async_send(void *arg)
244{
245 async_send_arg_t *a = (async_send_arg_t *)arg;
246
247 httpd_ws_frame_t ws_pkt = {
248 .type = HTTPD_WS_TYPE_TEXT,
249 .payload = (uint8_t *)a->data,
250 .len = a->len,
251 };
252
253 esp_err_t ret = httpd_ws_send_frame_async(a->hd, a->fd, &ws_pkt);
254 if (ret != ESP_OK) {
255 ESP_LOGW(TAG, "Async send failed to fd=%d: %d", a->fd, ret);
256 }
257
258 free(a->data);
259 free(a);
260}
261
262static void cleanup_server_init(ws_server_t *server, bool stop_httpd)
263{
264 g_server = NULL;
265 g_message_callback = NULL;
266 if (stop_httpd && server->server) {
267 httpd_stop(server->server);
268 server->server = NULL;
269 }
270 if (server->lock) {
271 vSemaphoreDelete(server->lock);
272 server->lock = NULL;
273 }
274}
275
276esp_err_t ws_server_init(ws_server_t *server, uint16_t port, ws_message_cb_t on_message)
277{
278 if (server->server != NULL) {
279 ESP_LOGE(TAG, "Server already initialized, call ws_server_stop first");
280 return ESP_ERR_INVALID_STATE;
281 }
282
283 memset(server, 0, sizeof(ws_server_t));
284 server->lock = xSemaphoreCreateMutex();
285 if (!server->lock) {
286 return ESP_ERR_NO_MEM;
287 }
288
289 g_server = server;
290 g_message_callback = on_message;
291
292 httpd_config_t config = HTTPD_DEFAULT_CONFIG();
293 config.server_port = port;
294 config.ctrl_port = port + 1;
295 config.max_open_sockets = WS_MAX_CONNECTIONS;
296 config.backlog_conn = WS_MAX_CONNECTIONS;
297 config.lru_purge_enable = true;
298 config.recv_wait_timeout = 3;
299 config.send_wait_timeout = 3;
300 config.keep_alive_enable = true;
301 config.keep_alive_idle = 5;
302 config.keep_alive_interval = 1;
303 config.keep_alive_count = 3;
304 config.stack_size = 12288;
305 config.open_fn = on_open;
306 config.close_fn = on_close;
307
308 esp_err_t ret = httpd_start(&server->server, &config);
309 if (ret != ESP_OK) {
310 ESP_LOGE(TAG, "Failed to start server: %d", ret);
311 cleanup_server_init(server, false);
312 return ret;
313 }
314
315 httpd_uri_t ws_uri = {
316 .uri = "/",
317 .method = HTTP_GET,
318 .handler = ws_handler,
319 .user_ctx = NULL,
320 .is_websocket = true,
321 .handle_ws_control_frames = true,
322 };
323
324 ret = httpd_register_uri_handler(server->server, &ws_uri);
325 if (ret != ESP_OK) {
326 ESP_LOGE(TAG, "Failed to register WS handler: %d", ret);
327 cleanup_server_init(server, true);
328 return ret;
329 }
330
331 httpd_uri_t options_uri = {
332 .uri = "/",
333 .method = HTTP_OPTIONS,
334 .handler = relay_nip11_options_handler,
335 .user_ctx = NULL,
336 };
337
338 ret = httpd_register_uri_handler(server->server, &options_uri);
339 if (ret != ESP_OK) {
340 ESP_LOGE(TAG, "Failed to register OPTIONS handler: %d", ret);
341 }
342
343 ESP_LOGI(TAG, "WebSocket server started on port %d", port);
344 return ESP_OK;
345}
346
347void ws_server_stop(ws_server_t *server)
348{
349 g_server = NULL;
350 g_message_callback = NULL;
351 g_disconnect_callback = NULL;
352
353 if (server->server) {
354 httpd_stop(server->server);
355 server->server = NULL;
356 }
357 if (server->lock) {
358 vSemaphoreDelete(server->lock);
359 server->lock = NULL;
360 }
361 memset(server->connections, 0, sizeof(server->connections));
362 server->connection_count = 0;
363}
364
365bool ws_server_is_running(ws_server_t *server)
366{
367 return server && server->server != NULL;
368}
369
370esp_err_t ws_server_send(ws_server_t *server, int fd, const char *data, size_t len)
371{
372 if (!server->server) return ESP_ERR_INVALID_STATE;
373
374 if (g_current_req && httpd_req_to_sockfd(g_current_req) == fd) {
375 httpd_ws_frame_t ws_pkt = {
376 .type = HTTPD_WS_TYPE_TEXT,
377 .payload = (uint8_t *)data,
378 .len = len,
379 };
380 return httpd_ws_send_frame(g_current_req, &ws_pkt);
381 }
382
383 async_send_arg_t *arg = malloc(sizeof(async_send_arg_t));
384 if (!arg) return ESP_ERR_NO_MEM;
385
386 arg->data = malloc(len);
387 if (!arg->data) {
388 free(arg);
389 return ESP_ERR_NO_MEM;
390 }
391
392 memcpy(arg->data, data, len);
393 arg->hd = server->server;
394 arg->fd = fd;
395 arg->len = len;
396
397 esp_err_t ret = httpd_queue_work(server->server, ws_async_send, arg);
398 if (ret != ESP_OK) {
399 free(arg->data);
400 free(arg);
401 return ret;
402 }
403 return ESP_OK;
404}
405
406esp_err_t ws_server_broadcast(ws_server_t *server, const char *data, size_t len)
407{
408 xSemaphoreTake(server->lock, portMAX_DELAY);
409
410 for (int i = 0; i < WS_MAX_CONNECTIONS; i++) {
411 if (server->connections[i].active) {
412 ws_server_send(server, server->connections[i].fd, data, len);
413 }
414 }
415
416 xSemaphoreGive(server->lock);
417 return ESP_OK;
418}
419
420void ws_server_close_connection(ws_server_t *server, int fd)
421{
422 if (!server || !server->server) {
423 return;
424 }
425 httpd_sess_trigger_close(server->server, fd);
426}
diff --git a/components/wisp_relay/ws_server.h b/components/wisp_relay/ws_server.h
new file mode 100644
index 0000000..4fe616e
--- /dev/null
+++ b/components/wisp_relay/ws_server.h
@@ -0,0 +1,41 @@
1#ifndef WS_SERVER_H
2#define WS_SERVER_H
3
4#include <stdint.h>
5#include <stdbool.h>
6#include <stddef.h>
7#include "esp_http_server.h"
8#include "freertos/FreeRTOS.h"
9#include "freertos/semphr.h"
10
11#define WS_MAX_CONNECTIONS 8
12#define WS_MAX_FRAME_SIZE 65536
13#define WS_IP_ADDR_MAX_LEN 48
14
15typedef struct {
16 int fd;
17 bool active;
18 uint32_t connected_at;
19 uint32_t last_activity;
20 char remote_ip[WS_IP_ADDR_MAX_LEN];
21} ws_connection_t;
22
23typedef struct {
24 httpd_handle_t server;
25 ws_connection_t connections[WS_MAX_CONNECTIONS];
26 SemaphoreHandle_t lock;
27 uint8_t connection_count;
28} ws_server_t;
29
30typedef void (*ws_message_cb_t)(int fd, const char *data, size_t len);
31typedef void (*ws_disconnect_cb_t)(int fd);
32
33esp_err_t ws_server_init(ws_server_t *server, uint16_t port, ws_message_cb_t on_message);
34void ws_server_set_disconnect_cb(ws_disconnect_cb_t cb);
35void ws_server_stop(ws_server_t *server);
36bool ws_server_is_running(ws_server_t *server);
37esp_err_t ws_server_send(ws_server_t *server, int fd, const char *data, size_t len);
38esp_err_t ws_server_broadcast(ws_server_t *server, const char *data, size_t len);
39void ws_server_close_connection(ws_server_t *server, int fd);
40
41#endif
generated by cgit v1.2.3 (git 2.39.1) at 2026-06-05 00:16:58 +0000