add new opcode + firehose change

5 files changed, 106 insertions, 165 deletions

diff --git a/src/common.c b/src/common.c
index c85d166..e59281d 100644
--- a/src/common.c
+++ b/src/common.c
@@ -25,67 +25,9 @@ typedef byte recipient[ 64 ];
 
 const uint16_t default_port = 8040;
 
-enum client_input {
-    // ask server for its information (responds with audit_response)
-    audit_server = 0x10,
-
-    // ask server for a random set of neighbors
-    // (input uint8_t neighbor_count, responds with neighbors_info)
-    get_neighbors = 0x11,
-
-    // monitor incoming packets
-    // (input struct firehose_request, responds with firehose_reply)
-    firehose = 0x12,
-
-    // give a packet to the relay
-    // (input struct data_packet, responds with push_packet_result)
-    push_packet = 0x13,
-
-    // get a packet from the server
-    // (input hash, responds with get_packet_result)
-    get_packet = 0x14,
-
-    // register a new neighbor relay to this server
-    // (input struct relay_neighbor, responds with register_neighbor_result)
-    register_neighbor = 0x15,
-
-    // query packets for a specific author
-    // (input public_key, pub_sign_key, u16 count, u64 offset)
-    // (responds with queried_author_packets)
-    query_packets = 0x16,
-
-    // query packets for a specific recipient
-    // note: to prevent DoS, you must provide a signature by the requested author to be approved.
-    // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
-    // offset`.
-    // server must verify the signing key is of the recipient's pubkey and signing key by looking at
-    // the packet's metadata.
-    // (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
-    // (responds with queried_author_packets)
-    query_by_recipient = 0x17,
-};
+char* global_packets_dir_path = "packets/";
 
-enum server_reply {
-    // pretty self-explanatory.
-    error = 0x1,
-
-    // type `struct relay_info`
-    audit_response = 0x18,
-    // type `struct relay_neighbor[ neighbor_count ]`
-    neighbors_info = 0x19,
-    // type `uint64_t new_offset, struct data_packet[ count ]`
-    firehose_reply = 0x1a,
-    // type `bool result_code`
-    push_packet_result = 0x1b,
-    // type `struct data_packet`
-    get_packet_result = 0x1c,
-    // type `bool result_code`
-    register_neighbor_result = 0x1d,
-    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
-    queried_author_packets = 0x1e,
-    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
-    queried_recipient_packets = 0x1f,
-};
+#include "req_res_codes.h"
 
 typedef uint8_t get_neighbors_request;
 
diff --git a/src/external.h b/src/external.h
index 13f65dd..aa6efc3 100644
--- a/src/external.h
+++ b/src/external.h
@@ -1,8 +1,8 @@
 #pragma once
 //
+#include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
-#include <stdbool.h>
 //
 
 typedef uint8_t byte;
@@ -18,72 +18,7 @@ typedef byte recipient[ 64 ];
 
 const uint16_t default_port = 8040;
 
-enum client_input {
-    // ask server for its information (responds with audit_response)
-    audit_server = 0x10,
-
-    // ask server for a random set of neighbors
-    // (input uint8_t neighbor_count, responds with neighbors_info)
-    get_neighbors = 0x11,
-
-    // monitor incoming packets
-    // (input struct firehose_request, responds with firehose_reply)
-    firehose = 0x12,
-
-    // give a packet to the relay
-    // (input struct data_packet, responds with push_packet_result)
-    push_packet = 0x13,
-
-    // get a packet from the server
-    // (input hash, responds with get_packet_result)
-    get_packet = 0x14,
-
-    // register a new neighbor relay to this server
-    // (input struct relay_neighbor, responds with register_neighbor_result)
-    register_neighbor = 0x15,
-
-    // query packets for a specific author
-    // note: to prevent DoS, you must provide a signature by the requested author to be approved.
-    // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
-    // offset`.
-    // server must verify the signing key is of the author's pubkey and signing key by looking at
-    // the packet's metadata.
-    // (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
-    // (responds with queried_author_packets)
-    query_packets = 0x16,
-
-    // query packets for a specific recipient
-    // note: to prevent DoS, you must provide a signature by the requested author to be approved.
-    // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
-    // offset`.
-    // server must verify the signing key is of the recipient's pubkey and signing key by looking at
-    // the packet's metadata.
-    // (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
-    // (responds with queried_author_packets)
-    query_by_recipient = 0x17,
-};
-
-enum server_reply {
-    // pretty self-explanatory.
-    error = 0x1,
-
-    // type `struct relay_info`
-    audit_response = 0x18,
-    // type `struct relay_neighbor[ neighbor_count ]`
-    neighbors_info = 0x19,
-    // type `uint64_t new_offset, struct data_packet[ count ]`
-    firehose_reply = 0x1a,
-    // type `bool result_code`
-    push_packet_result = 0x1b,
-    // type `struct data_packet`
-    get_packet_result = 0x1c,
-    // type `bool result_code`
-    register_neighbor_result = 0x1d,
-    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
-    queried_author_packets = 0x1e,
-    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
-    queried_recipient_packets = 0x1f,
-};
+#include "req_res_codes.h"
 
 typedef uint8_t get_neighbors_request;
 
@@ -139,7 +74,7 @@ struct data_packet {
         public_key author;             // message author
         public_key author_signing_pub; // message signing pubkey
         uint8_t    recipient_count;    // number of recipients. leave as 0 for public.
-        recipient recipients[ 32 ];   // up to 32 recipients for packet
+        recipient  recipients[ 32 ];   // up to 32 recipients for packet
         byte       data[ 65536 ];      // packet data
         signature  signature;          // data signature
     } contents;
diff --git a/src/firehose.c b/src/firehose.c
index 6a4a134..c5d232d 100644
--- a/src/firehose.c
+++ b/src/firehose.c
@@ -6,7 +6,7 @@
 #include <stdio.h>
 #include <string.h>
 
-#define FIREHOSE_BUFFER_SIZE 1 << 14
+#define FIREHOSE_BUFFER_SIZE 1 << 15
 
 typedef struct {
     hash     packet_hash;
@@ -14,10 +14,10 @@ typedef struct {
 } water_splash;
 
 typedef struct {
-    water_splash    entries[ FIREHOSE_BUFFER_SIZE ];
+    water_splash*   entries;
     uint16_t        head;
     uint16_t        tail;
-    uint16_t        count;
+    uint32_t        count;
     uint64_t        total_packets;
     pthread_mutex_t mutex;
 } fire_buffer;
@@ -30,8 +30,10 @@ uint16_t firehose_get_count(void);
 uint64_t firehose_get_total_packets(void);
 uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* output_entries);
 
+void* alloc_map(uint64_t size);
+
 void firehose_init(void) {
-    memset(firehose_buffer.entries, 0, sizeof(firehose_buffer.entries));
+    firehose_buffer.entries       = alloc_map(sizeof(water_splash) * FIREHOSE_BUFFER_SIZE);
     firehose_buffer.head          = 0;
     firehose_buffer.tail          = 0;
     firehose_buffer.count         = 0;
@@ -80,7 +82,6 @@ uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* out
     }
     uint16_t start_index = (firehose_buffer.head + relative_offset) % FIREHOSE_BUFFER_SIZE;
 
-    pthread_mutex_lock(&packet_storage_mutex);
     for (int i = 0; i < count; i++) {
         if (relative_offset + i >= firehose_buffer.count) { break; }
 
@@ -89,8 +90,7 @@ uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* out
             &output_entries[ i ], &firehose_buffer.entries[ current_index ], sizeof(water_splash));
         copied_count++;
     }
-    pthread_mutex_unlock(&packet_storage_mutex);
 
     pthread_mutex_unlock(&firehose_buffer.mutex);
     return copied_count;
-}
+}
\ No newline at end of file
diff --git a/src/req_res_codes.h b/src/req_res_codes.h
new file mode 100644
index 0000000..3903401
--- /dev/null
+++ b/src/req_res_codes.h
@@ -0,0 +1,69 @@
+enum client_input {
+    // ask server for its information (responds with audit_response)
+    audit_server = 0x10,
+
+    // ask server for a random set of neighbors
+    // (input uint8_t neighbor_count, responds with neighbors_info)
+    get_neighbors = 0x11,
+
+    // monitor incoming packets
+    // (input struct firehose_request, responds with firehose_reply)
+    firehose = 0x12,
+
+    // give a packet to the relay
+    // (input struct data_packet, responds with push_packet_result)
+    push_packet = 0x13,
+
+    // get a packet from the server
+    // (input hash, responds with get_packet_result)
+    get_packet = 0x14,
+
+    // register a new neighbor relay to this server
+    // (input struct relay_neighbor, responds with register_neighbor_result)
+    register_neighbor = 0x15,
+
+    // query packets for a specific author
+    // (input public_key, pub_sign_key, u16 count, u64 offset)
+    // (responds with queried_author_packets)
+    query_packets = 0x16,
+
+    // query packets for a specific recipient
+    // note: to prevent DoS, you must provide a signature by the requested author to be approved.
+    // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
+    // offset`.
+    // server must verify the signing key is of the recipient's pubkey and signing key by looking at
+    // the packet's metadata.
+    // (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
+    // (responds with queried_author_packets)
+    query_by_recipient = 0x17,
+
+    // query packets by in-memory id and hash.
+    // this is to be used in tandem with firehose, in an effort to reduce traffic usage.
+    // note that this only works for entries currently loaded into memory. (meaning the id can be
+    // invalidated at any time)
+    // (input hash, u64 id)
+    // (responds with get_packet_result)
+    query_by_id_hash = 0x18
+};
+
+enum server_reply {
+    // pretty self-explanatory.
+    error = 0x1,
+
+    // type `struct relay_info`
+    audit_response = 0x19,
+    // type `struct relay_neighbor[ neighbor_count ]`
+    neighbors_info = 0x1a,
+    // type `uint64_t new_offset, struct data_packet[ count ]`
+    firehose_reply = 0x1b,
+    // type `bool result_code`
+    push_packet_result = 0x1c,
+    // type `struct data_packet`
+    get_packet_result = 0x1d,
+    // type `bool result_code`
+    register_neighbor_result = 0x1e,
+    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
+    queried_author_packets = 0x1f,
+    // type `u16 count, u64 new_offset, struct data_packet[ count ]`
+    queried_recipient_packets = 0x20,
+};
\ No newline at end of file
diff --git a/src/server.c b/src/server.c
index cc1f45d..3140788 100644
--- a/src/server.c
+++ b/src/server.c
@@ -464,7 +464,7 @@ void paste_server(
             if (unlikely(req.count > 64)) req.count = 64;
 
             log(_info,
-                "<%.*s> request type: firehose count %i offset %li",
+                "<%.*s> request type: firehose count %i offset %lu",
                 (int) ip_size,
                 ip_addr,
                 req.count,
@@ -473,38 +473,20 @@ void paste_server(
             water_splash entries[ req.count ];
             uint16_t     copied_count = firehose_get_packets(req.offset, req.count, entries);
 
-            size_t total_packets_size = PACKET_SIZE * copied_count;
-
-            size_t output_len  = 1 + 8 + total_packets_size;
+            size_t output_len  = 1 + 8 + (sizeof(water_splash) * copied_count);
             byte*  output_data = calloc(output_len, sizeof(byte));
             output_data[ 0 ]   = firehose_reply;
             size_t offset      = 1;
 
-            uint64_t new_offset     = req.offset + copied_count;
-            output_data[ offset++ ] = new_offset & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 8) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 16) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 24) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 32) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 40) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 48) & 0xFF;
-            output_data[ offset++ ] = (new_offset >> 56) & 0xFF;
+            uint64_t new_offset = req.offset + copied_count;
+            write_u64_le(output_data + offset, new_offset);
+            offset += 8;
 
             for (int i = 0; i < copied_count; i++) {
-                struct data_packet_entry* entry = &packet_index[ entries[ i ].storage_index ];
-                byte                      decompressed[ sizeof(struct data_packet) ];
-                uLongf                    decompressed_len = sizeof(struct data_packet);
-                if (uncompress(
-                        decompressed,
-                        &decompressed_len,
-                        entry->compressed_payload,
-                        entry->compressed_size)
-                    != Z_OK) {
-                    log(_error, "firehose: decompression failed");
-                    continue;
-                }
-                data_packet2s(*(struct data_packet*) decompressed, output_data + offset);
-                offset += PACKET_SIZE;
+                memcpy(output_data + offset, &entries[ i ].storage_index, 8);
+                offset += 8;
+                memcpy(output_data + offset, entries[ i ].packet_hash, 32);
+                offset += 32;
             }
 
             reply(output_len, output_data);
@@ -540,7 +522,12 @@ void paste_server(
                 for (int j = 0; j < 8; j++) {
                     uint16_t word
                         = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ];
-                    offset += sprintf(ip_string + offset, "%x%s", word, (j < 7) ? ":" : "");
+                    offset += snprintf(
+                        ip_string + offset,
+                        sizeof(ip_string) - offset,
+                        "%x%s",
+                        word,
+                        (j < 7) ? ":" : "");
                 }
             }
 
@@ -821,8 +808,10 @@ void save_packet(const struct data_packet packet) {
     }
 }
 
+char* global_neighbors_save_path = "neighbors.bin";
+
 void save_neighbors() {
-    FILE* file = fopen("neighbors.bin", "wb");
+    FILE* file = fopen(global_neighbors_save_path, "wb");
     if (!file) {
         perror("failed to save neighbors");
         return;
@@ -843,7 +832,7 @@ void save_neighbors() {
 }
 
 void load_neighbors() {
-    FILE* file = fopen("neighbors.bin", "rb");
+    FILE* file = fopen(global_neighbors_save_path, "rb");
     if (!file) { return; }
 
     size_t count = 0;
@@ -1065,6 +1054,12 @@ int main(int argc, char** argv) {
         free(port);
     }
 
+    char* packets_path = get_config("packets_dir", false);
+    if (packets_path) global_packets_dir_path = packets_path;
+
+    char* neighbors_save_path = get_config("neighbors_path", false);
+    if (neighbors_save_path) global_neighbors_save_path = neighbors_save_path;
+
     fprintf(
         stderr,
         "-- moonpaste server v%i.%i --\n"