stuff

1 files changed, 284 insertions, 25 deletions

diff --git a/src/packet_storage.c b/src/packet_storage.c
index 52a80d0..569eed4 100644
--- a/src/packet_storage.c
+++ b/src/packet_storage.c
@@ -8,8 +8,9 @@
 #include <sys/stat.h>
 #include <zlib.h>
 
-const uint64_t max_storage = 1 << 16;
-const uint64_t max_authors = 1 << 14;
+const uint64_t max_storage    = 1 << 16;
+const uint64_t max_authors    = 1 << 14;
+const uint64_t max_recipients = 1 << 14;
 
 #define hash_fmt                                                                                   \
     "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x"                             \
@@ -31,6 +32,8 @@ struct data_packet_entry {
     uint64_t prev;
     uint64_t author_next;
     uint64_t author_prev;
+    uint64_t recipient_next;
+    uint64_t recipient_prev;
     bool     is_full;
 }       *packet_index;
 byte    *packet_arena;
@@ -44,7 +47,16 @@ struct author_entry {
     bool       is_full;
 } *author_storage;
 
-pthread_mutex_t author_storage_mutex = PTHREAD_MUTEX_INITIALIZER;
+struct recipient_entry {
+    public_key recipient;
+    uint64_t   packet_head;
+    uint64_t   next;
+    uint64_t   prev;
+    bool       is_full;
+} *recipient_storage;
+
+pthread_mutex_t author_storage_mutex    = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t recipient_storage_mutex = PTHREAD_MUTEX_INITIALIZER;
 
 #define invalid_index ((uint64_t) 0)
 
@@ -142,6 +154,23 @@ void save_author_packet_hash(const public_key author, const hash packet_hash) {
     fclose(f);
 }
 
+void save_recipient_packet_hash(const public_key recipient, const hash packet_hash) {
+    const char *dir = "packets_by_recipient";
+    mkdir(dir, 0755);
+
+    char recipient_hex[ 65 ];
+    sprintf(recipient_hex, hash_fmt, hash_ref(recipient));
+
+    char filename[ 256 ];
+    snprintf(filename, sizeof(filename), "%s/%s.list", dir, recipient_hex);
+
+    FILE *f = fopen(filename, "ab");
+    if (!f) return;
+    fwrite(packet_hash, 32, 1, f);
+    fputc('\n', f);
+    fclose(f);
+}
+
 uint64_t get_author_entry(const public_key auth_key, bool create) {
     uint64_t hash_val;
     memcpy(&hash_val, auth_key, sizeof(uint64_t));
@@ -198,6 +227,62 @@ uint64_t get_author_entry(const public_key auth_key, bool create) {
     return invalid_index;
 }
 
+uint64_t get_recipient_entry(const public_key recip_key, bool create) {
+    uint64_t hash_val;
+    memcpy(&hash_val, recip_key, sizeof(uint64_t));
+    uint64_t start_index = hash_val % max_recipients;
+    uint64_t cur_index   = start_index;
+
+    pthread_mutex_lock(&recipient_storage_mutex);
+
+    struct recipient_entry *current = &recipient_storage[ cur_index ];
+    while (true) {
+        if (current->is_full && memcmp(current->recipient, recip_key, 32) == 0) {
+            pthread_mutex_unlock(&recipient_storage_mutex);
+            return cur_index;
+        }
+        if (current->next == invalid_index) break;
+        cur_index = current->next;
+        current   = &recipient_storage[ cur_index ];
+    }
+
+    if (!create) {
+        pthread_mutex_unlock(&recipient_storage_mutex);
+        return invalid_index;
+    }
+
+    if (!current->is_full) {
+        memcpy(current->recipient, recip_key, 32);
+        current->packet_head = invalid_index;
+        current->is_full     = true;
+        current->next        = invalid_index;
+        current->prev        = invalid_index;
+        pthread_mutex_unlock(&recipient_storage_mutex);
+        return cur_index;
+    }
+
+    uint64_t probes = 0;
+    uint64_t lp_idx = (cur_index + 1) % max_recipients;
+    while (probes < max_recipients) {
+        struct recipient_entry *empty = &recipient_storage[ lp_idx ];
+        if (!empty->is_full) {
+            memcpy(empty->recipient, recip_key, 32);
+            empty->packet_head = invalid_index;
+            empty->is_full     = true;
+            empty->next        = invalid_index;
+            empty->prev        = cur_index;
+            current->next      = lp_idx;
+            pthread_mutex_unlock(&recipient_storage_mutex);
+            return lp_idx;
+        }
+        lp_idx = (lp_idx + 1) % max_recipients;
+        probes++;
+    }
+
+    pthread_mutex_unlock(&recipient_storage_mutex);
+    return invalid_index;
+}
+
 struct data_packet *load_packet(FILE *file) {
     if (!file) return NULL;
 
@@ -269,6 +354,9 @@ uint64_t set_data_packet(struct data_packet *new_packet) {
             current_entry->compressed_size    = compressed_len;
             pthread_mutex_unlock(&packet_storage_mutex);
             save_author_packet_hash(new_packet->contents.author, new_packet->hash);
+            for (uint8_t i = 0; i < new_packet->contents.recipient_count; i++) {
+                save_recipient_packet_hash(new_packet->contents.recipients[ i ], new_packet->hash);
+            }
             return cur_index;
         }
         if (current_entry->next == invalid_index) break;
@@ -282,35 +370,65 @@ uint64_t set_data_packet(struct data_packet *new_packet) {
         current_entry->is_full            = true;
         current_entry->next               = invalid_index;
         current_entry->prev               = invalid_index;
+        current_entry->author_next        = invalid_index;
+        current_entry->author_prev        = invalid_index;
+        current_entry->recipient_next     = invalid_index;
+        current_entry->recipient_prev     = invalid_index;
         packet_count++;
-        pthread_mutex_unlock(&packet_storage_mutex);
-        save_author_packet_hash(new_packet->contents.author, new_packet->hash);
-        return cur_index;
+    } else {
+        uint64_t probes             = 0;
+        uint64_t linear_probe_index = (cur_index + 1) % max_storage;
+        while (probes < max_storage) {
+            struct data_packet_entry *empty_slot = &packet_index[ linear_probe_index ];
+            if (!empty_slot->is_full) {
+                empty_slot->compressed_payload = compressed_payload;
+                empty_slot->compressed_size    = compressed_len;
+                empty_slot->is_full            = true;
+                empty_slot->next               = invalid_index;
+                empty_slot->prev               = cur_index;
+                empty_slot->author_next        = invalid_index;
+                empty_slot->author_prev        = invalid_index;
+                empty_slot->recipient_next     = invalid_index;
+                empty_slot->recipient_prev     = invalid_index;
+                current_entry->next            = linear_probe_index;
+                packet_count++;
+                cur_index = linear_probe_index;
+                break;
+            }
+            linear_probe_index = (linear_probe_index + 1) % max_storage;
+            probes++;
+        }
+        if (probes >= max_storage) {
+            pthread_mutex_unlock(&packet_storage_mutex);
+            arena_free(compressed_payload);
+            return invalid_index;
+        }
     }
 
-    uint64_t probes             = 0;
-    uint64_t linear_probe_index = (cur_index + 1) % max_storage;
-    while (probes < max_storage) {
-        struct data_packet_entry *empty_slot = &packet_index[ linear_probe_index ];
-        if (!empty_slot->is_full) {
-            empty_slot->compressed_payload = compressed_payload;
-            empty_slot->compressed_size    = compressed_len;
-            empty_slot->is_full            = true;
-            empty_slot->next               = invalid_index;
-            empty_slot->prev               = cur_index;
-            current_entry->next            = linear_probe_index;
-            packet_count++;
-            pthread_mutex_unlock(&packet_storage_mutex);
-            save_author_packet_hash(new_packet->contents.author, new_packet->hash);
-            return linear_probe_index;
+    uint64_t auth_idx = get_author_entry(new_packet->contents.author, true);
+    if (auth_idx != invalid_index) {
+        current_entry->author_next = author_storage[ auth_idx ].packet_head;
+        if (current_entry->author_next != invalid_index) {
+            packet_index[ current_entry->author_next ].author_prev = cur_index;
+        }
+        author_storage[ auth_idx ].packet_head = cur_index;
+    }
+
+    for (uint8_t i = 0; i < new_packet->contents.recipient_count; i++) {
+        uint64_t recip_idx = get_recipient_entry(new_packet->contents.recipients[ i ], true);
+        if (recip_idx != invalid_index) {
+            current_entry->recipient_next = recipient_storage[ recip_idx ].packet_head;
+            if (current_entry->recipient_next != invalid_index) {
+                packet_index[ current_entry->recipient_next ].recipient_prev = cur_index;
+            }
+            recipient_storage[ recip_idx ].packet_head = cur_index;
+            save_recipient_packet_hash(new_packet->contents.recipients[ i ], new_packet->hash);
         }
-        linear_probe_index = (linear_probe_index + 1) % max_storage;
-        probes++;
     }
 
     pthread_mutex_unlock(&packet_storage_mutex);
-    arena_free(compressed_payload);
-    return invalid_index;
+    save_author_packet_hash(new_packet->contents.author, new_packet->hash);
+    return cur_index;
 }
 
 struct data_packet_entry *get_data_packet(const hash _hash) {
@@ -384,6 +502,10 @@ struct data_packet_entry *get_data_packet(const hash _hash) {
             temp_entry.is_full         = true;
             temp_entry.next            = invalid_index;
             temp_entry.prev            = invalid_index;
+            temp_entry.author_next     = invalid_index;
+            temp_entry.author_prev     = invalid_index;
+            temp_entry.recipient_next  = invalid_index;
+            temp_entry.recipient_prev  = invalid_index;
             free(packet);
             return &temp_entry;
         }
@@ -450,6 +572,26 @@ bool _delete_at_index(uint64_t index) {
     if (a_prev != invalid_index) packet_index[ a_prev ].author_next = a_next;
     if (a_next != invalid_index) packet_index[ a_next ].author_prev = a_prev;
 
+    for (uint8_t i = 0;
+         i < ((struct data_packet *) current_entry->compressed_payload)->contents.recipient_count;
+         i++) {
+        public_key recip_key;
+        memcpy(
+            recip_key,
+            ((struct data_packet *) current_entry->compressed_payload)->contents.recipients[ i ],
+            32);
+        uint64_t recip_idx = get_recipient_entry(recip_key, false);
+        if (recip_idx != invalid_index) {
+            if (recipient_storage[ recip_idx ].packet_head == index) {
+                recipient_storage[ recip_idx ].packet_head = current_entry->recipient_next;
+            }
+        }
+        uint64_t r_prev = current_entry->recipient_prev;
+        uint64_t r_next = current_entry->recipient_next;
+        if (r_prev != invalid_index) packet_index[ r_prev ].recipient_next = r_next;
+        if (r_next != invalid_index) packet_index[ r_next ].recipient_prev = r_prev;
+    }
+
     if (current_entry->prev != invalid_index) {
         packet_index[ current_entry->prev ].next = current_entry->next;
     }
@@ -463,6 +605,10 @@ bool _delete_at_index(uint64_t index) {
     current_entry->prev               = invalid_index;
     current_entry->compressed_payload = NULL;
     current_entry->compressed_size    = 0;
+    current_entry->author_next        = invalid_index;
+    current_entry->author_prev        = invalid_index;
+    current_entry->recipient_next     = invalid_index;
+    current_entry->recipient_prev     = invalid_index;
     return true;
 }
 
@@ -583,4 +729,117 @@ uint16_t get_packets_for_author(
 
     free(all_packets);
     return collected;
+}
+
+uint16_t get_packets_for_recipient(
+    const public_key    recipient,
+    uint64_t            offset,
+    uint16_t            max_count,
+    struct data_packet *out_packets,
+    uint64_t           *out_new_offset) {
+    uint64_t recip_idx = get_recipient_entry(recipient, false);
+    if (recip_idx == invalid_index) return 0;
+
+    uint64_t fetch_factor = ((offset + (uint64_t) max_count) * 5LLU) / 2;
+
+    pthread_mutex_lock(&packet_storage_mutex);
+    uint64_t curr      = recipient_storage[ recip_idx ].packet_head;
+    uint16_t mem_count = 0;
+    while (curr != invalid_index) {
+        mem_count++;
+        curr = packet_index[ curr ].recipient_next;
+    }
+
+    struct data_packet *all_packets
+        = malloc((mem_count + fetch_factor) * sizeof(struct data_packet));
+    if (!all_packets) {
+        pthread_mutex_unlock(&packet_storage_mutex);
+        return 0;
+    }
+    size_t all_count = 0;
+
+    curr = recipient_storage[ recip_idx ].packet_head;
+    while (curr != invalid_index) {
+        byte   decompressed[ sizeof(struct data_packet) ];
+        uLongf decompressed_len = sizeof(struct data_packet);
+        if (uncompress(
+                decompressed,
+                &decompressed_len,
+                packet_index[ curr ].compressed_payload,
+                packet_index[ curr ].compressed_size)
+            != Z_OK) {
+            printf("get_packets_for_recipient: decompression failed\n");
+            continue;
+        }
+        memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_packet));
+        curr = packet_index[ curr ].recipient_next;
+    }
+    pthread_mutex_unlock(&packet_storage_mutex);
+
+    char recipient_hex[ 65 ];
+    sprintf(recipient_hex, hash_fmt, hash_ref(recipient));
+
+    char filename[ 256 ];
+    snprintf(filename, sizeof(filename), "packets_by_recipient/%s.list", recipient_hex);
+
+    FILE  *f          = fopen(filename, "rb");
+    size_t disk_added = 0;
+    if (f) {
+        fseek(f, 0, SEEK_END);
+        long fsize = ftell(f);
+        fseek(f, 0, SEEK_SET);
+        uint64_t total = fsize / 33;
+
+        uint64_t fetch_count = fetch_factor;
+        if (fetch_count > total) fetch_count = total;
+
+        for (uint64_t i = 0; i < fetch_count; i++) {
+            long pos = (total - 1 - i) * 33;
+            fseek(f, pos, SEEK_SET);
+
+            hash phash;
+            if (fread(phash, 32, 1, f) != 1) break;
+            if (fgetc(f) != '\n') break;
+
+            bool dupe = false;
+            for (size_t j = 0; j < all_count; j++) {
+                if (memcmp(all_packets[ j ].hash, phash, 32) == 0) {
+                    dupe = true;
+                    break;
+                }
+            }
+            if (dupe) continue;
+
+            struct data_packet_entry *ent = get_data_packet(phash);
+            if (!ent) continue;
+
+            byte   decompressed[ sizeof(struct data_packet) ];
+            uLongf decompressed_len = sizeof(struct data_packet);
+            if (uncompress(
+                    decompressed, &decompressed_len, ent->compressed_payload, ent->compressed_size)
+                != Z_OK) {
+                printf("get_packets_for_recipient: decompression failed for disk packet\n");
+                continue;
+            }
+            memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_packet));
+            disk_added++;
+        }
+        fclose(f);
+    }
+
+    qsort(all_packets, all_count, sizeof(struct data_packet), get_packets_for_author__cmp);
+
+    if (offset >= all_count) {
+        free(all_packets);
+        *out_new_offset = offset;
+        return 0;
+    }
+
+    size_t   avail     = all_count - offset;
+    uint16_t collected = (avail > max_count) ? max_count : avail;
+    memcpy(out_packets, all_packets + offset, collected * sizeof(struct data_packet));
+    *out_new_offset = offset + collected;
+
+    free(all_packets);
+    return collected;
 }
\ No newline at end of file