#include <dirent.h>
#include <fcntl.h>
#include <pthread.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
// dependency: zlib
#include <zlib.h>
//
#include "common.c"
#include "config.c"
#include "firehose.c"
#include "omen.c"
#include "packet_storage.c"
#include "plugin.c"
#include "security.c"
#include "sha256.c"

uint16_t current_port = default_port;

const uint64_t max_neighbors = 1 << 10;

uint64_t               current_neighbor_ptr = 0;
uint64_t               neighbor_count       = 0;
struct relay_neighbor* neighbor_storage;
bool*                  empty_neighbor_slots;
pthread_mutex_t        neighbor_storage_mutex = PTHREAD_MUTEX_INITIALIZER;

struct server_stats {
    byte     major_version;
    byte     minor_version;
    byte     motd[ 128 ];
    uint16_t retention_hours;
    byte     leading_zeros;
    bool     accepts_plain;
} server_stats = {
    .major_version   = 0,
    .minor_version   = 1,
    .motd            = "A moonpaste server.",
    .retention_hours = 168,
    .leading_zeros   = 12,
    .accepts_plain   = true,
};

bool add_neighbor(struct relay_neighbor new_neighbor) {
    if (decide_neighbor_registry_func) {
        if (!decide_neighbor_registry_func(new_neighbor)) return false;
    }

    log(_info, "adding neighbor");

    pthread_mutex_lock(&neighbor_storage_mutex);
    if (neighbor_count >= max_neighbors) {
        pthread_mutex_unlock(&neighbor_storage_mutex);
        return false;
    }

    uint64_t add_idx = -1;
    for (uint64_t i = 0; i < max_neighbors; i++) {
        if (empty_neighbor_slots[ i ]) {
            add_idx = i;
            break;
        }
    }

    if (add_idx != -1) {
        neighbor_storage[ add_idx ]     = new_neighbor;
        empty_neighbor_slots[ add_idx ] = false;
        neighbor_count++;
        pthread_mutex_unlock(&neighbor_storage_mutex);
        return true;
    }

    pthread_mutex_unlock(&neighbor_storage_mutex);
    return false;
}

void* propagate_packet(void* _packet) {
    struct data_packet packet = *(struct data_packet*) _packet;

    if (decide_packet_propogation_func) {
        if (!decide_packet_propogation_func(packet)) return NULL;
    }

    log(_debug, "propagating packet " hash_fmt, hash_ref(packet.hash));

    size_t send_len = 1 + sizeof(struct data_packet);
    byte   send_data[ send_len * sizeof(byte) ];
    send_data[ 0 ] = push_packet;
    data_packet2s(packet, send_data + 1);

    pthread_mutex_lock(&neighbor_storage_mutex);
    for (uint64_t i = 0; i < neighbor_count; i++) {
        struct relay_neighbor neighbor = neighbor_storage[ i ];
        if (neighbor.active) {
            byte*  response_data = NULL;
            size_t response_len  = 0;
            int    result        = -1;

            if (neighbor.has_ipv4) {
                char ip_string[ 16 ];
                sprintf(
                    ip_string,
                    "%d.%d.%d.%d",
                    neighbor.v4_addr[ 0 ],
                    neighbor.v4_addr[ 1 ],
                    neighbor.v4_addr[ 2 ],
                    neighbor.v4_addr[ 3 ]);

                result = study(
                    ip_string, neighbor.port, send_data, send_len, &response_data, &response_len);
            } else if (neighbor.has_ipv6) {
                char ip_string[ 46 ];
                int  offset = 0;
                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) ? ":" : "");
                }

                result = study(
                    ip_string, neighbor.port, send_data, send_len, &response_data, &response_len);
            }

            if (result != 0) { log(_error, "failed to send packet to neighbor"); }

            if (response_data) { free(response_data); }
        }
    }
    pthread_mutex_unlock(&neighbor_storage_mutex);

    free(_packet);
    return NULL;
}

void save_packet(const struct data_packet packet);

#define reply(__len, ...)                                                                          \
    ({                                                                                             \
        *out     = (byte*) __VA_ARGS__;                                                            \
        *out_len = __len;                                                                          \
    })

void paste_server(
    char* ip_addr, size_t ip_size, byte* in, size_t in_len, byte** out, size_t* out_len) {
    log(_info, "<%.*s> incoming request with size %li", (int) ip_size, ip_addr, in_len);

    if (paste_server_start_func) {
        paste_server_start_func(&ip_addr, &ip_size, &in, &in_len, &out, &out_len);
    }

    if (in_len < 8) {
        reply(17, strdup("\x01input too small"));
        return;
    }

    enum client_input input_type = in[ 0 ];

    switch (input_type) {
        case audit_server: {
            log(_info, "<%.*s> request type: audit", (int) ip_size, ip_addr);

            byte* output_data = calloc(1 + 237, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }
            output_data[ 0 ] = audit_response;

            struct relay_info info = { 0 };

            pthread_mutex_lock(&neighbor_storage_mutex);
            pthread_mutex_lock(&packet_storage_mutex);

            info.major_version                = server_stats.major_version;
            info.minor_version                = server_stats.minor_version;
            info.capabilities.retention_hours = server_stats.retention_hours;
            info.capabilities.can_store       = packet_count < max_storage - 1;
            info.capabilities.leading_zeros   = server_stats.leading_zeros;
            info.capabilities.accepts_plain   = server_stats.accepts_plain;

            info.statistics.neighbors    = neighbor_count;
            info.statistics.held_packets = packet_count;

            for (int i = 0; i < 4; ++i) {
                info.statistics.quick_connect[ i ] = neighbor_storage[ i ];
            }

            memcpy(info.motd, server_stats.motd, 128);

            if (modify_server_stats_func) { modify_server_stats_func(&info); }

            pthread_mutex_unlock(&packet_storage_mutex);
            pthread_mutex_unlock(&neighbor_storage_mutex);

            relay_info2s(info, output_data + 1);

            reply(1 + 245, output_data);
            return;
        }

        case get_neighbors: {
            byte count = in[ 1 ];
            log(_info,
                "<%.*s> request type: get neighbors, count %i",
                (int) ip_size,
                ip_addr,
                count);

            size_t output_len  = 1 + (RELAY_NEIGHBOR_WIRE_LEN * count);
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }
            output_data[ 0 ] = neighbors_info;

            size_t offset = 1;

            for (byte i = 0; i < count; i++) {
                relay_neighbor2s(neighbor_storage[ i ], output_data + offset);
                offset += RELAY_NEIGHBOR_WIRE_LEN;
            }

            reply(output_len, output_data);
            return;
        }

        case get_packet: {
            if (in_len < 1 + 32) {
                byte* output_data = calloc(1 + 17, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "input too small", 17);
                reply(18, output_data);
                return;
            }

            hash packet_hash;
            bool success;
            s2hash(32, in + 1, &success, packet_hash);

            log(_info,
                "<%.*s> request type: get packet with hash " hash_fmt,
                (int) ip_size,
                ip_addr,
                hash_ref(packet_hash));

            if (!success) {
                byte* output_data = calloc(1 + 12, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "invalid hash", 12);
                reply(13, output_data);
                return;
            }

            struct data_packet_entry* entry = get_data_packet(packet_hash);
            if (entry == NULL) {
                byte* output_data = calloc(1 + 14, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "packet not found", 14);
                reply(15, output_data);
                return;
            }

            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) {
                byte* output_data = calloc(1 + 22, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "decompression failed", 22);
                reply(23, output_data);
                return;
            }

            size_t output_len  = 1 + PACKET_SIZE;
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }

            output_data[ 0 ] = get_packet_result;
            data_packet2s(*(struct data_packet*) decompressed, output_data + 1);

            reply(output_len, output_data);
            return;
        }

        case push_packet: {
            bool               success;
            struct data_packet packet = s2data_packet(in_len - 1, in + 1, &success);

            log(_info,
                "<%.*s> request type: push packet with hash " hash_fmt,
                (int) ip_size,
                ip_addr,
                hash_ref(packet.hash));

            size_t output_len  = 1 + 1;
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) { return; }
            output_data[ 0 ] = push_packet_result;

            if (!success) {
                log(_warn, "<%.*s> malformed data", (int) ip_size, ip_addr);
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            if (modify_packet_func) { modify_packet_func(&packet); }

            if (!check_leading_zeros(packet.hash, server_stats.leading_zeros)) {
                log(_warn, "bad leads");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            const size_t hash_input_len = 4 + CONTENTS_SIZE;
            byte         hash_input_buffer[ hash_input_len ];
            write_u32_le(hash_input_buffer, packet.nonce);
            byte* contents_buffer = hash_input_buffer + 4;

            contents_buffer[ 0 ] = packet.contents.encrypted ? 1 : 0;
            write_u64_le(contents_buffer + 1, packet.contents.created_time_ms);
            memcpy(contents_buffer + 9, packet.contents.author, 32);
            memcpy(contents_buffer + 41, packet.contents.author_signing_pub, 32);
            contents_buffer[ 73 ] = packet.contents.recipient_count;
            memcpy(contents_buffer + 74, packet.contents.recipients, 32 * MAX_RECIPIENTS);
            memcpy(contents_buffer + 74 + 32 * MAX_RECIPIENTS, packet.contents.data, 65536);
            memcpy(
                contents_buffer + 74 + 32 * MAX_RECIPIENTS + 65536, packet.contents.signature, 64);

            struct sha256 calculated_hash_struct = sha256_hash(hash_input_buffer, hash_input_len);
            if (memcmp(calculated_hash_struct.hash, packet.hash, sizeof(hash)) != 0) {
                log(_warn,
                    "bad hash " hash_fmt " vs " hash_fmt,
                    hash_ref(calculated_hash_struct.hash),
                    hash_ref(packet.hash));
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            uint64_t current_time = get_current_time_ms();
            if (packet.contents.created_time_ms > current_time) {
                log(_warn, "packet timestamp is in the future");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            if (current_time - packet.contents.created_time_ms
                > server_stats.retention_hours * (60 * 60 * 1000)) {
                log(_warn, "packet expired");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            if (!packet.contents.encrypted && !server_stats.accepts_plain) {
                log(_warn, "plaintext not accepted by server");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            bool sig_ok = false;
            if (!packet.contents.encrypted) {
                size_t msg_len = CONTENTS_SIZE - sizeof(signature);
                sig_ok         = verify_raw_signature(
                    contents_buffer,
                    msg_len,
                    packet.contents.signature,
                    packet.contents.author_signing_pub);
            } else {
                secure_payload sp;
                memcpy(sp.nonce, packet.contents.data, sizeof(sp.nonce));
                memcpy(
                    sp.encrypted_data,
                    packet.contents.data + sizeof(sp.nonce),
                    sizeof(sp.encrypted_data));
                memcpy(sp.signature, packet.contents.signature, sizeof(sp.signature));
                sig_ok = verify_signature(sp, packet.contents.author_signing_pub);
            }

            if (!sig_ok) {
                log(_warn, "bad signature");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            uint64_t storage_index = set_data_packet(&packet);
            firehose_push(packet.hash, storage_index);

            if (storage_index != invalid_index) {
                output_data[ 1 ] = true;

                save_packet(packet);

                if (((float) rand() / (float) RAND_MAX) > 0.5) {
                    struct data_packet* thread_arg = malloc(sizeof(struct data_packet));
                    if (thread_arg) {
                        *thread_arg = packet;
                        pthread_t prop_thread;
                        pthread_create(&prop_thread, NULL, propagate_packet, thread_arg);
                        pthread_detach(prop_thread);
                    }
                }
            } else {
                output_data[ 1 ] = false;
            }

            reply(output_len, output_data);
            return;
        }

        case firehose: {
            if (in_len < 1 + 9) {
                byte* output_data = calloc(1 + 17, sizeof(byte));
                output_data[ 0 ]  = error;
                memcpy(output_data + 1, "input too small", 17);
                reply(18, output_data);
                return;
            }
            struct firehose_request req;
            bool                    success;
            req = s2firehose_request(in_len - 1, in + 1, &success);

            if (!success) {
                byte* output_data = calloc(1 + 14, sizeof(byte));
                output_data[ 0 ]  = error;
                memcpy(output_data + 1, "invalid request", 14);
                reply(15, output_data);
                return;
            }

            if (unlikely(req.count > 64)) req.count = 64;

            log(_info,
                "<%.*s> request type: firehose count %i offset %li",
                (int) ip_size,
                ip_addr,
                req.count,
                req.offset);

            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;
            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;

            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;
            }

            reply(output_len, output_data);
            return;
        }

        case register_neighbor: {
            bool                  success;
            struct relay_neighbor neighbor = s2relay_neighbor(in_len - 1, in + 1, &success);

            if (!success) {
                byte* output_data = calloc(1 + 14, sizeof(byte));
                output_data[ 0 ]  = error;
                memcpy(output_data + 1, "invalid request", 14);
                reply(15, output_data);
                return;
            }

            log(_info, "<%.*s> request type: neighbor registration", (int) ip_size, ip_addr);

            char ip_string[ INET6_ADDRSTRLEN ];

            if (neighbor.has_ipv4) {
                sprintf(
                    ip_string,
                    "%d.%d.%d.%d",
                    neighbor.v4_addr[ 0 ],
                    neighbor.v4_addr[ 1 ],
                    neighbor.v4_addr[ 2 ],
                    neighbor.v4_addr[ 3 ]);
            } else if (neighbor.has_ipv6) {
                int offset = 0;
                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) ? ":" : "");
                }
            }

            struct relay_info audit_result = run_server_audit(ip_string, neighbor.port);
            if (!audit_result.capabilities.can_store && audit_result.major_version == 0
                && audit_result.minor_version == 0) {
                byte* err_reply = malloc(2);
                if (err_reply) {
                    err_reply[ 0 ] = error;
                    err_reply[ 1 ] = 0;
                }
                reply(2, err_reply);
                break;
            }

            byte* __reply = malloc(2);
            if (__reply) {
                __reply[ 0 ] = register_neighbor_result;
                __reply[ 1 ] = add_neighbor(neighbor);
            }
            reply(2, __reply);
        } break;

        case query_packets: {
            if (in_len != 1 + 32 + 32 + 64 + 2 + 8) {
                *out     = NULL;
                *out_len = 0;
                return;
            }
            public_key author_enc, author_sign;
            signature  sig;
            size_t     off = 1;
            memcpy(author_enc, in + off, 32);
            off += 32;
            memcpy(author_sign, in + off, 32);
            off += 32;
            memcpy(sig, in + off, 64);
            off += 64;
            uint16_t req_count = read_u16_le(in + off);
            off += 2;
            uint64_t req_offset = read_u64_le(in + off);
            off += 8;
            byte msg[ 32 + 32 + 2 + 8 ];
            memcpy(msg, author_enc, 32);
            memcpy(msg + 32, author_sign, 32);
            write_u16_le(msg + 64, req_count);
            write_u64_le(msg + 66, req_offset);
            if (!verify_raw_signature(msg, sizeof(msg), sig, author_sign)) {
                *out     = NULL;
                *out_len = 0;
                return;
            }
            if (req_count > 128) req_count = 128;
            struct data_packet* packets = malloc((size_t) req_count * sizeof(struct data_packet));
            if (!packets) {
                *out     = NULL;
                *out_len = 0;
                return;
            }
            uint64_t new_offset;
            uint16_t actual_count
                = get_packets_for_author(author_enc, req_offset, req_count, packets, &new_offset);
            uint16_t verified_count = 0;
            for (uint16_t i = 0; i < actual_count; i++) {
                if (memcmp(packets[ i ].contents.author_signing_pub, author_sign, 32) == 0) {
                    packets[ verified_count++ ] = packets[ i ];
                }
            }
            size_t _out_len = 1 + 2 + 8 + (size_t) verified_count * PACKET_SIZE;
            byte*  _out     = malloc(_out_len);
            if (!_out) {
                free(packets);
                *out     = NULL;
                *out_len = 0;
                return;
            }
            off           = 0;
            _out[ off++ ] = queried_author_packets;
            write_u16_le(_out + off, verified_count);
            off += 2;
            write_u64_le(_out + off, new_offset);
            off += 8;
            for (uint16_t i = 0; i < verified_count; i++) {
                data_packet2s(packets[ i ], _out + off);
                off += PACKET_SIZE;
            }
            free(packets);
            *out     = _out;
            *out_len = _out_len;
            return;
        } break;

        default: {
            log(_info, "<%.*s> request type: invalid method", (int) ip_size, ip_addr);

            byte* output_data = calloc(1 + 14, sizeof(byte));
            output_data[ 0 ]  = error;
            memcpy(output_data + 1, "invalid input", 14);
            reply(15, output_data);
            return;
        }
    }
}

bool try_hugepages = true;

void* alloc_map(uint64_t size) {
    static const byte RW = PROT_READ | PROT_WRITE;
    static const byte PA = MAP_PRIVATE | MAP_ANONYMOUS;

    void* alloc;
    if (try_hugepages) {
        alloc = mmap(NULL, size, RW, PA | MAP_HUGE_2MB, -1, 0);
        if (alloc == MAP_FAILED) { goto NO_HP; }
    } else {
    NO_HP:
        alloc = mmap(NULL, size, RW, PA, -1, 0);
        if (alloc != MAP_FAILED) {
            printf("[note] no hugepages support present");
            try_hugepages = false;
        }
    }

    if (alloc == MAP_FAILED) return NULL;
    return alloc;
}

pthread_t     packet_cleanup_thread;
volatile bool packet_cleanup_thread_running = true;

pthread_t     disk_persistence_thread;
volatile bool disk_persistence_thread_running = true;

pthread_t     neighbor_check_thread;
volatile bool neighbor_check_thread_running = true;

bool file_exists(const char* filename) { return access(filename, F_OK) == 0; }

void save_packet(const struct data_packet packet) {
    if (decide_packet_save_to_disk_func) {
        if (!decide_packet_save_to_disk_func(packet)) return;
    }

    log(_debug, "saving packet " hash_fmt " to disk", hash_ref(packet.hash));

    mkdir("packets", 0755);

    char filename_tmp[ 256 ];
    char filename_final[ 256 ];
    snprintf(
        filename_tmp, sizeof(filename_tmp), "packets/" hash_fmt ".bin.tmp", hash_ref(packet.hash));
    snprintf(
        filename_final, sizeof(filename_final), "packets/" hash_fmt ".bin", hash_ref(packet.hash));

    byte buffer[ PACKET_SIZE ];
    data_packet2s(packet, buffer);

    uLongf comp_bound = compressBound(PACKET_SIZE);
    byte   comp_buf[ comp_bound ];

    uLongf comp_len = comp_bound;
    int    zres     = compress2(comp_buf, &comp_len, buffer, PACKET_SIZE, Z_BEST_COMPRESSION);
    if (zres != Z_OK) {
        log(_error, "save_packet: compress2 failed (%d)", zres);
        return;
    }

    FILE* f = fopen(filename_tmp, "wb");
    if (!f) {
        perror("save_packet: fopen tmp");
        return;
    }

    if (fwrite(comp_buf, 1, comp_len, f) != comp_len) {
        perror("save_packet: write failed");
        fclose(f);
        unlink(filename_tmp);
        return;
    }

    fflush(f);
#if defined(__unix__) || defined(__APPLE__)
    int fd = fileno(f);
    if (fd >= 0) fsync(fd);
#endif
    fclose(f);

    if (rename(filename_tmp, filename_final) != 0) {
        perror("save_packet: rename failed");
        unlink(filename_tmp);
    }
}

void save_neighbors() {
    FILE* file = fopen("neighbors.bin", "wb");
    if (!file) {
        perror("failed to save neighbors");
        return;
    }

    pthread_mutex_lock(&neighbor_storage_mutex);
    for (uint64_t i = 0; i < max_neighbors; i++) {
        if (!empty_neighbor_slots[ i ]) {
            byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ];
            relay_neighbor2s(neighbor_storage[ i ], buffer);
            if (fwrite(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) != RELAY_NEIGHBOR_WIRE_LEN) {
                perror("failed to write neighbor entry");
            }
        }
    }
    pthread_mutex_unlock(&neighbor_storage_mutex);
    fclose(file);
}

void load_neighbors() {
    FILE* file = fopen("neighbors.bin", "rb");
    if (!file) { return; }

    size_t count = 0;

    byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ];
    while (fread(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) == RELAY_NEIGHBOR_WIRE_LEN) {
        bool                  success;
        struct relay_neighbor neighbor
            = s2relay_neighbor(RELAY_NEIGHBOR_WIRE_LEN, buffer, &success);
        if (success) {
            count++;
            add_neighbor(neighbor);
        }
    }
    fclose(file);

    log(_info, "loaded %li neighbors", count);
}

void* fs_persistence(void* arg) {
    while (disk_persistence_thread_running) {
        sleep(2 * 60);
        log(_info, "saving neighbors...");
        save_neighbors();
    }

    return NULL;
}

void* cleanup_packets(void* arg) {
    while (packet_cleanup_thread_running) {
        sleep(30 * 60);

        log(_warn, "running cleanup");
        uint64_t current_time_ms = get_current_time_ms();
        uint64_t retention_ms    = (uint64_t) server_stats.retention_hours * 60 * 60 * 1000;

        for (uint64_t i = 0; i < max_storage; i++) {
            pthread_mutex_lock(&packet_storage_mutex);

            if (!packet_index[ i ].is_full) {
                pthread_mutex_unlock(&packet_storage_mutex);
                continue;
            }

            byte   decompressed[ sizeof(struct data_packet) ];
            uLongf decompressed_len = sizeof(struct data_packet);
            if (uncompress(
                    decompressed,
                    &decompressed_len,
                    packet_index[ i ].compressed_payload,
                    packet_index[ i ].compressed_size)
                != Z_OK) {
                log(_error, "cleanup_packets: decompression failed");
                pthread_mutex_unlock(&packet_storage_mutex);
                continue;
            }
            struct data_packet* packet = (struct data_packet*) decompressed;

            bool result = decide_packet_cleanup_func
                              ? decide_packet_cleanup_func(*packet)
                              : (current_time_ms - packet->contents.created_time_ms > retention_ms);

            if (result) {
                log(_debug, "cleaning up old packet hashed " hash_fmt, hash_ref(packet->hash));

                char filename[ 128 ];
                sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(packet->hash));

                if (remove(filename) != 0) log(_error, "failed to delete packet %s", filename);

                _delete_at_index(i);
            }

            pthread_mutex_unlock(&packet_storage_mutex);
        }

        system("find packets/ -mtime +7 -delete");

        DIR* dir = opendir("packets");
        if (dir) {
            struct dirent* entry;
            while ((entry = readdir(dir)) != NULL) {
                if (entry->d_type == DT_REG) {
                    char filepath[ 256 ];
                    snprintf(filepath, sizeof(filepath), "packets/%s", entry->d_name);

                    FILE* fd = decide_packet_cleanup_func ? fopen(filepath, "rb") : NULL;
                    struct data_packet* new = decide_packet_cleanup_func ? load_packet(fd) : NULL;

                    struct stat st;
                    bool        result
                        = decide_packet_cleanup_func
                              ? decide_packet_cleanup_func(*new)
                              : (stat(filepath, &st) == 0
                                 && current_time_ms - (uint64_t) st.st_mtime * 1000 > retention_ms);

                    if (decide_packet_cleanup_func) {
                        fclose(fd);
                        free(new);
                    }

                    if (result) {
                        if (remove(filepath) != 0) log(_error, "failed to delete %s", filepath);
                    }
                }
            }
            closedir(dir);
        }
    }

    return NULL;
}

void* check_neighbors(void* arg) {
    sleep(5 * 30);

    while (neighbor_check_thread_running) {
        pthread_mutex_lock(&neighbor_storage_mutex);
        for (uint64_t i = 0; i < max_neighbors; i++) {
            if (!empty_neighbor_slots[ i ]) {
                struct relay_neighbor neighbor = neighbor_storage[ i ];

                byte*  response_data = NULL;
                size_t response_len  = 0;
                int    result        = -1;

                byte audit_request[ 1 ] = { audit_server };

                if (neighbor.has_ipv4) {
                    char ip_string[ 16 ];
                    sprintf(
                        ip_string,
                        "%d.%d.%d.%d",
                        neighbor.v4_addr[ 0 ],
                        neighbor.v4_addr[ 1 ],
                        neighbor.v4_addr[ 2 ],
                        neighbor.v4_addr[ 3 ]);
                    result = study(
                        ip_string, neighbor.port, audit_request, 1, &response_data, &response_len);
                } else if (neighbor.has_ipv6) {
                    char ip_string[ 46 ];
                    int  offset = 0;

                    for (int j = 0; j < 8; j++) {
                        uint16_t word
                            = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ];
                        offset += snprintf(
                            ip_string + offset,
                            sizeof(ip_string) - offset,
                            "%x%s",
                            word,
                            (j < 7) ? ":" : "");
                    }

                    result = study(
                        ip_string, neighbor.port, audit_request, 1, &response_data, &response_len);
                }

                if (result != 0 || response_data[ 0 ] != audit_response) {
                    empty_neighbor_slots[ i ] = true;
                    neighbor_count--;
                    log(_warn, "removed dead neighbor");
                }

                if (response_data) { free(response_data); }
            }
        }
        pthread_mutex_unlock(&neighbor_storage_mutex);

        sleep(60 * 5);
    }
    return NULL;
}

int main(int argc, char** argv) {
    parse_arguments(argc, argv);

    char* __log_level = get_config("log", false);
    if (__log_level) {
        if (isdigit(*__log_level)) {
            int _log_level = atoi(__log_level);
            if (_log_level < 1 || _log_level > 4) {
                fprintf(
                    stderr,
                    "invalid log level %d (1 = debug, 2 = info, 3 = warn, 4 = error)\n",
                    _log_level);
                exit(1);
            }
            log_level = _log_level;
        } else {
            if (strcmp(__log_level, "debug") == 0) log_level = 1;
            else if (strcmp(__log_level, "info") == 0)
                log_level = 2;
            else if (strcmp(__log_level, "warn") == 0)
                log_level = 3;
            else if (strcmp(__log_level, "error") == 0)
                log_level = 4;
            else {
                fprintf(
                    stderr,
                    "invalid log level %s (1 = debug, 2 = info, 3 = warn, 4 = error)\n",
                    __log_level);
                exit(1);
            }
        }

        free(__log_level);
    }

    char* port = get_config("port", false);
    if (port) {
        if (!isdigit(*port)) {
            fprintf(stderr, "invalid port value '%s', must be a number\n", port);
            exit(1);
        }

        current_port = atoi(port);
        free(port);
    }

    fprintf(
        stderr,
        "-- moonpaste server v%i.%i --\n"
        "port %i\n\n",
        server_stats.major_version,
        server_stats.minor_version,
        current_port);

    packet_index = alloc_map(max_storage * sizeof(struct data_packet_entry));
    packet_arena_size
        = max_storage * (sizeof(uint8_t) + sizeof(uint32_t) + sizeof(struct data_packet));
    packet_arena   = alloc_map(packet_arena_size);
    author_storage = alloc_map(max_authors * sizeof(struct author_entry));

    neighbor_storage     = alloc_map(max_neighbors * sizeof(struct relay_neighbor));
    empty_neighbor_slots = alloc_map(max_neighbors * sizeof(bool));
    memset(empty_neighbor_slots, 1, max_neighbors * sizeof(bool));

    if (!packet_index || !packet_arena || !author_storage || !neighbor_storage
        || !empty_neighbor_slots) {
        perror("init alloc failed");
        return 1;
    }

    if (pthread_create(&packet_cleanup_thread, NULL, cleanup_packets, NULL) != 0) {
        perror("failed to create packet cleanup thread");
        return 1;
    }

    load_neighbors();

    if (pthread_create(&disk_persistence_thread, NULL, fs_persistence, NULL) != 0) {
        perror("failed to create persistence thread");
        return 1;
    }

    if (pthread_create(&neighbor_check_thread, NULL, check_neighbors, NULL) != 0) {
        perror("failed to create neighbor audit thread");
        return 1;
    }

    firehose_init();

    char* plugin_path = get_config("plugin", false);
    if (plugin_path) {
        load_plugins(plugin_path);
        free(plugin_path);
    }

    omen(current_port, paste_server);

    packet_cleanup_thread_running = false;
    pthread_join(packet_cleanup_thread, NULL);

    return 0;
}