//
// Created by WolverinDEV on 04/05/2020.
//

#include <event.h>
#include <cassert>
#include <netinet/tcp.h>
#include <log/LogUtils.h>
#include <misc/net.h>
#include "./LocalFileProvider.h"
#include "./duration_utils.h"
#include "LocalFileProvider.h"

#if defined(TCP_CORK) && !defined(TCP_NOPUSH)
    #define TCP_NOPUSH TCP_CORK
#endif

using namespace ts::server::file;
using namespace ts::server::file::transfer;

inline void add_network_event(FileClient& transfer, event* ev, bool& ev_throttle_readd_flag, bool ignore_bandwidth) {
    timeval tv{0, 1}, *ptv{nullptr};
    {
        auto timeout = transfer.networking.disconnect_timeout;
        if(timeout.time_since_epoch().count() > 0) {
            auto now = std::chrono::system_clock::now();
            if(now < timeout) {
                auto duration = timeout - now;

                auto seconds = std::chrono::duration_cast<std::chrono::seconds>(timeout - now);
                duration -= seconds;

                auto microseconds = std::chrono::duration_cast<std::chrono::microseconds>(timeout - now);

                tv.tv_sec = seconds.count();
                tv.tv_usec = microseconds.count();
            }
            ptv = &tv;
        }
    }

    if(!ignore_bandwidth) {
        if(ev_throttle_readd_flag) return; /* we're already throttled */

        timeval ttv{};
        if(transfer.networking.throttle.should_throttle(ttv)) {
            if(transfer.networking.event_throttle)
                event_add(transfer.networking.event_throttle, &ttv);
            ev_throttle_readd_flag = true;
            return;
        }
    }

    event_add(ev, ptv);
}

void FileClient::add_network_read_event(bool ignore_bandwidth) {
    std::shared_lock slock{this->state_mutex};

    switch (this->state) {
        case STATE_DISCONNECTING:
        case STATE_DISCONNECTED:
            return;

        case STATE_AWAITING_KEY:
        case STATE_TRANSFERRING:
            break;

        default:
            assert(false);
            return;
    }
    if(this->state != STATE_AWAITING_KEY && this->state != STATE_TRANSFERRING)
        return;

    add_network_event(*this, this->networking.event_read, this->networking.add_event_read, ignore_bandwidth);
}

void FileClient::add_network_write_event(bool ignore_bandwidth) {
    std::shared_lock slock{this->state_mutex};
    this->add_network_write_event_nolock(ignore_bandwidth);
}

void FileClient::add_network_write_event_nolock(bool ignore_bandwidth) {
    switch (this->state) {
        case STATE_DISCONNECTED:
            return;

        case STATE_DISCONNECTING:
            if(this->transfer->direction == Transfer::DIRECTION_UPLOAD)
                return;
            /* flush our write buffer */
            break;

        case STATE_TRANSFERRING:
            break;

        //case STATE_PENDING:
        //case STATE_AWAITING_KEY:
        default:
            assert(false);
            break;
    }

    add_network_event(*this, this->networking.event_write, this->networking.add_event_write, ignore_bandwidth);
}

bool FileClient::send_file_bytes(const void *snd_buffer, size_t size) {
    if(this->networking.protocol == FileClient::PROTOCOL_TS_V1) {
        auto tbuffer = allocate_buffer(size);
        tbuffer->length = size;
        tbuffer->offset = 0;
        memcpy(tbuffer->data, snd_buffer, size);

        size_t buffer_size;
        {
            std::lock_guard block{this->buffer.mutex};
            *this->buffer.buffer_tail = tbuffer;
            this->buffer.buffer_tail = &tbuffer->next;

            buffer_size = (this->buffer.bytes += size);
        }

        this->add_network_write_event(false);
        return buffer_size > TRANSFER_MAX_CACHED_BYTES;
    } else if(this->networking.protocol == FileClient::PROTOCOL_HTTPS) {
        this->networking.pipe_ssl.send(pipes::buffer_view{snd_buffer, size});
        return this->buffer.bytes > TRANSFER_MAX_CACHED_BYTES;
    } else {
        return false;
    }
}

NetworkingStartResult LocalFileTransfer::start_networking() {
    assert(!this->network.active);

    this->network.active = true;
    this->network.event_base = event_base_new();
    if(!this->network.event_base) return NetworkingStartResult::OUT_OF_MEMORY;

    bool bound{false};
    for(auto& binding : this->network.bindings) {
        binding->file_descriptor = socket(binding->address.ss_family, SOCK_STREAM | SOCK_NONBLOCK, 0);
        if(!binding->file_descriptor) {
            logWarning(LOG_FT, "Failed to allocate socket for {}: {}/{}", binding->hostname, errno, strerror(errno));
            continue;
        }


        int enable = 1, disabled = 0;

        if (setsockopt(binding->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int)) < 0)
            logWarning(LOG_FT, "Failed to activate SO_REUSEADDR for binding {} ({} | {})", binding->hostname, errno, strerror(errno));

        if(setsockopt(binding->file_descriptor, IPPROTO_TCP, TCP_NOPUSH, &disabled, sizeof disabled) < 0)
            logWarning(LOG_FT, "Failed to deactivate TCP_NOPUSH for binding {} ({} | {})", binding->hostname, errno, strerror(errno));

        if(binding->address.ss_family == AF_INET6) {
            if(setsockopt(binding->file_descriptor, IPPROTO_IPV6, IPV6_V6ONLY, &enable, sizeof(int)) < 0)
                logWarning(LOG_FT, "Failed to activate IPV6_V6ONLY for IPv6 binding {} ({} | {})", binding->hostname, errno, strerror(errno));
        }
        if(fcntl(binding->file_descriptor, F_SETFD, FD_CLOEXEC) < 0)
            logWarning(LOG_FT, "Failed to set flag FD_CLOEXEC for binding {} ({} | {})", binding->hostname, errno, strerror(errno));


        if (bind(binding->file_descriptor, (struct sockaddr *) &binding->address, sizeof(binding->address)) < 0) {
            logError(LOG_FT, "Failed to bind server to {}. (Failed to bind socket: {} | {})", binding->hostname, errno, strerror(errno));
            goto reset_binding;
        }

        if (listen(binding->file_descriptor, 8) < 0) {
            logError(LOG_FT, "Failed to bind server to {}. (Failed to listen: {} | {})", binding->hostname, errno, strerror(errno));
            goto reset_binding;
        }

        binding->handle = this;
        binding->accept_event = event_new(this->network.event_base, binding->file_descriptor, (unsigned) EV_READ | (unsigned) EV_PERSIST, &LocalFileTransfer::callback_transfer_network_accept, &*binding);
        if(!binding->accept_event)
            goto reset_binding;

        event_add(binding->accept_event, nullptr);
        logMessage(LOG_FT, "Started to listen on {}:{}", binding->hostname, net::port(binding->address));

        bound = true;
        continue;

        reset_binding:
        if(binding->accept_event) {
            event_free(binding->accept_event);
            binding->accept_event = nullptr;
        }

        if(binding->file_descriptor > 0)
            ::close(binding->file_descriptor);
        binding->file_descriptor = 0;

        binding->handle = nullptr;
    }
    if(!bound) {
        event_base_free(std::exchange(this->network.event_base, nullptr));
        return NetworkingStartResult::NO_BINDINGS;
    }

    this->network.dispatch_thread = std::thread(&LocalFileTransfer::dispatch_loop_network, this);
    return NetworkingStartResult::SUCCESS;
}

void LocalFileTransfer::shutdown_networking() {
    if(!this->network.active) return;
    this->network.active = false;

    for(auto& binding : this->network.bindings) {
        if(binding->accept_event) {
            event_del_block(binding->accept_event);
            event_free(binding->accept_event);
            binding->accept_event = nullptr;
        }

        if(binding->file_descriptor > 0)
            ::close(binding->file_descriptor);
        binding->file_descriptor = 0;

        binding->handle = nullptr;
    }

    {
        std::unique_lock tlock{this->transfers_mutex};
        auto transfers = this->transfers_;
        tlock.unlock();
        for(const auto& transfer : transfers) {
            std::unique_lock slock{transfer->state_mutex};
            this->disconnect_client(transfer, slock, false);
        }
    }

    auto ev_base = std::exchange(this->network.event_base, nullptr);
    if(ev_base)
        event_base_loopbreak(ev_base);

    if(this->network.dispatch_thread.joinable())
        this->network.dispatch_thread.join();

    if(ev_base)
        event_base_free(ev_base);
}

void LocalFileTransfer::dispatch_loop_network(void *provider_ptr) {
    auto provider = reinterpret_cast<LocalFileTransfer*>(provider_ptr);

    while(provider->network.active) {
        assert(provider->network.event_base);
        event_base_loop(provider->network.event_base, EVLOOP_NO_EXIT_ON_EMPTY);
    }
}

NetworkInitializeResult LocalFileTransfer::initialize_networking(const std::shared_ptr<FileClient> &client, int file_descriptor) {
    client->networking.file_descriptor = file_descriptor;

    client->networking.event_read = event_new(this->network.event_base, file_descriptor, EV_READ, &LocalFileTransfer::callback_transfer_network_read, &*client);
    client->networking.event_write = event_new(this->network.event_base, file_descriptor, EV_WRITE, &LocalFileTransfer::callback_transfer_network_write, &*client);
    client->networking.event_throttle = evtimer_new(this->network.event_base, &LocalFileTransfer::callback_transfer_network_throttle, &*client);

    if(!client->networking.event_read || !client->networking.event_write || !client->networking.event_throttle)
        goto oom_exit;

    client->add_network_read_event(true);

    client->timings.connected = std::chrono::system_clock::now();
    client->timings.last_write = client->timings.connected;
    client->timings.last_read = client->timings.connected;

    return NetworkInitializeResult::SUCCESS;

    oom_exit:
    if(auto event{std::exchange(client->networking.event_read, nullptr)}; event)
        event_free(event);
    if(auto event{std::exchange(client->networking.event_write, nullptr)}; event)
        event_free(event);
    if(auto event{std::exchange(client->networking.event_throttle, nullptr)}; event)
        event_free(event);

    return NetworkInitializeResult::OUT_OF_MEMORY;
}

void LocalFileTransfer::finalize_networking(const std::shared_ptr<FileClient> &client, std::unique_lock<std::shared_mutex>& state_lock) {
    assert(state_lock.owns_lock());

    auto ev_read = std::exchange(client->networking.event_read, nullptr);
    auto ev_write = std::exchange(client->networking.event_write, nullptr);
    auto ev_throttle = std::exchange(client->networking.event_throttle, nullptr);

    state_lock.unlock();
    if (ev_read) {
        event_del_block(ev_read);
        event_free(ev_read);
    }
    if (ev_write) {
        event_del_block(ev_write);
        event_free(ev_write);
    }
    if (ev_throttle) {
        event_del_block(ev_throttle);
        event_free(ev_throttle);
    }
    state_lock.lock();

    if (client->networking.file_descriptor > 0) {
        ::shutdown(client->networking.file_descriptor, SHUT_RDWR);
        ::close(client->networking.file_descriptor);
    }
    client->networking.file_descriptor = 0;
}

void LocalFileTransfer::callback_transfer_network_accept(int fd, short, void *ptr_binding) {
    auto binding = reinterpret_cast<NetworkBinding*>(ptr_binding);
    auto transfer = binding->handle;

    sockaddr_storage address{};
    socklen_t address_length{sizeof(address)};
    auto client_fd = ::accept4(fd, reinterpret_cast<sockaddr*>(&address), &address_length, SOCK_NONBLOCK);
    if(client_fd <= 0) {
        /* TODO: Reserve one file descriptor in case of out of file descriptors (see current implementation) */
        logError(LOG_FT, "Failed to accept new client: {}/{}", errno, strerror(errno));
        return;
    }

    auto client = std::make_shared<FileClient>(transfer);
    memcpy(&client->networking.address, &address, sizeof(sockaddr_storage));

    logMessage(LOG_FT, "{} Connection received.", client->log_prefix());
    auto ninit = transfer->initialize_networking(client, client_fd);
    switch(ninit) {
        case NetworkInitializeResult::SUCCESS: {
            std::lock_guard tlock{transfer->transfers_mutex};
            transfer->transfers_.push_back(std::move(client));
            return;
        }
        case NetworkInitializeResult::OUT_OF_MEMORY:
            client->state = FileClient::STATE_DISCONNECTED; /* required else the deallocate assert will fail */
            logError(LOG_FT, "{} Failed to initialize transfer client because we ran out of memory. Closing connection.",  client->log_prefix());
            ::close(client_fd);
            return;

        default:
            client->state = FileClient::STATE_DISCONNECTED; /* required else the deallocate assert will fail */
            logError(LOG_FT, "{} Failed to initialize transfer client. Closing connection.", client->log_prefix());
            ::close(client_fd);
            return;
    }
}

void LocalFileTransfer::callback_transfer_network_throttle(int, short, void *ptr_transfer) {
    auto transfer = reinterpret_cast<FileClient*>(ptr_transfer);

    if(std::exchange(transfer->networking.add_event_write, false))
        transfer->add_network_write_event(true);

    if(std::exchange(transfer->networking.add_event_read, false))
        transfer->add_network_read_event(true);
}

void LocalFileTransfer::callback_transfer_network_read(int fd, short events, void *ptr_transfer) {
    auto transfer = reinterpret_cast<FileClient*>(ptr_transfer);

    if((unsigned) events & (unsigned) EV_TIMEOUT) {
        /* should never happen, receive timeouts are done via the client tick */
    }

    if((unsigned) events & (unsigned) EV_READ) {
        constexpr size_t buffer_size{4096};
        char buffer[buffer_size];

        while(true) {
            const auto max_read_buffer = transfer->networking.throttle.bytes_left();
            if(!max_read_buffer) break; /* network throttle */

            auto read = ::recv(fd, buffer, std::min(buffer_size, std::min(max_read_buffer, transfer->networking.max_read_buffer_size)), MSG_NOSIGNAL | MSG_DONTWAIT);
            //logTrace(0, "Read {}, max {} | {}", read, std::min(buffer_size, std::min(max_read_buffer, transfer->networking.max_read_buffer_size)), max_read_buffer);
            if(read <= 0) {
                if(read == 0) {
                    std::unique_lock slock{transfer->state_mutex};
                    auto original_state = transfer->state;
                    transfer->handle->disconnect_client(transfer->shared_from_this(), slock, false);
                    slock.unlock();

                    switch(original_state) {
                        case FileClient::STATE_AWAITING_KEY:
                            logMessage(LOG_FT, "{} Disconnected without sending any key or initializing a transfer.", transfer->log_prefix());
                            break;
                        case FileClient::STATE_TRANSFERRING: {
                            assert(transfer->transfer);

                            /* since we're counting the bytes directly on read, a disconnect would mean that not all data has been transferred */
                            if(transfer->transfer->direction == Transfer::DIRECTION_UPLOAD) {
                                logMessage(LOG_FT, "{} Disconnected while receiving file. Received {} out of {} bytes",
                                        transfer->log_prefix(), transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset);
                            } else if(transfer->transfer->direction == Transfer::DIRECTION_DOWNLOAD) {
                                logMessage(LOG_FT, "{} Disconnected while sending file. Send {} out of {} bytes",
                                        transfer->log_prefix(), transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset);
                            }

                            transfer->handle->report_transfer_statistics(transfer->shared_from_this());
                            if(auto callback{transfer->handle->callback_transfer_aborted}; callback)
                                callback(transfer->transfer, { TransferError::UNEXPECTED_CLIENT_DISCONNECT, "" });
                            break;
                        }
                        default:
                            break;
                    }
                } else {
                    if(errno == EAGAIN) {
                        transfer->add_network_read_event(false);
                        return;
                    }

                    std::unique_lock slock{transfer->state_mutex};
                    auto original_state = transfer->state;
                    transfer->handle->disconnect_client(transfer->shared_from_this(), slock, false);
                    slock.unlock();

                    switch(original_state) {
                        case FileClient::STATE_AWAITING_KEY:
                            logMessage(LOG_FT, "{} Received a read error for an unauthenticated client: {}/{}. Closing connection.", transfer->log_prefix(), errno, strerror(errno));
                            break;
                        case FileClient::STATE_TRANSFERRING:
                            assert(transfer->transfer);

                            /* since we're counting the bytes directly on read, a disconnect would mean that not all data has been transferred */
                            if(transfer->transfer->direction == Transfer::DIRECTION_UPLOAD) {
                                logMessage(LOG_FT, "{} Received read error while receiving file. Received {} out of {} bytes: {}/{}",
                                        transfer->log_prefix(),  transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset, errno, strerror(errno));
                            } else if(transfer->transfer->direction == Transfer::DIRECTION_DOWNLOAD) {
                                logMessage(LOG_FT, "{} Received read error while sending file. Send {} out of {} bytes: {}/{}",
                                        transfer->log_prefix(),  transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset, errno, strerror(errno));
                            }

                            transfer->handle->report_transfer_statistics(transfer->shared_from_this());
                            if(auto callback{transfer->handle->callback_transfer_aborted}; callback)
                                callback(transfer->transfer, { TransferError::NETWORK_IO_ERROR, strerror(errno) });
                            break;
                        default:
                            break;
                    }
                    return;
                }
                return;
            } else {
                transfer->timings.last_read = std::chrono::system_clock::now();
                transfer->statistics.network_bytes_received += read;
                bool throttle_read = transfer->networking.throttle.increase_bytes(read);

                if(transfer->state != FileClient::STATE_AWAITING_KEY && !(transfer->state == FileClient::STATE_TRANSFERRING && transfer->transfer->direction == Transfer::DIRECTION_UPLOAD)) {
                    debugMessage(LOG_FT, "{} Received {} bytes without any need. Dropping them.", transfer->log_prefix(), read);
                    return;
                }

                size_t bytes_buffered{0};
                if(transfer->state == FileClient::STATE_AWAITING_KEY) {
                    bytes_buffered = transfer->handle->handle_transfer_read(transfer->shared_from_this(), buffer, read);
                } else if(transfer->state == FileClient::STATE_TRANSFERRING) {
                    if(transfer->transfer->direction == Transfer::DIRECTION_UPLOAD) {
                        bytes_buffered = transfer->handle->handle_transfer_read(transfer->shared_from_this(), buffer, read);
                    } else {
                        debugMessage(LOG_FT, "{} Received {} bytes without any need. Dropping them.", transfer->log_prefix(), read);
                    }
                }

                if(transfer->state == FileClient::STATE_DISCONNECTING || transfer->state == FileClient::STATE_DISCONNECTED)
                    break;

                if(bytes_buffered > TRANSFER_MAX_CACHED_BYTES) {
                    transfer->buffer.buffering_stopped = true;
                    debugMessage(LOG_FT, "{} Stopping network read, temp buffer full.", transfer->log_prefix());
                    return; /* no read event readd, buffer filled */
                }

                if(throttle_read)
                    break;
            }
        }

        transfer->add_network_read_event(false); /* read event is not persistent */
    }
}

void LocalFileTransfer::callback_transfer_network_write(int fd, short events, void *ptr_transfer) {
    auto transfer = reinterpret_cast<FileClient*>(ptr_transfer);

    if((unsigned) events & (unsigned) EV_TIMEOUT) {
        if(transfer->state == FileClient::STATE_DISCONNECTING) {
            {
                std::unique_lock slock{transfer->state_mutex};
                transfer->handle->disconnect_client(transfer->shared_from_this(), slock, false);
            }

            logMessage(LOG_FT, "{} Flush timeout. Force closing connection.", transfer->log_prefix());
            return;
        }
    }

    if((unsigned) events & (unsigned) EV_WRITE) {
        if(transfer->state != FileClient::STATE_DISCONNECTING && transfer->state != FileClient::STATE_TRANSFERRING) {
            debugMessage(LOG_FT, "{} Tried to write data to send only stream. Dropping buffers.", transfer->log_prefix());

            std::unique_lock block{transfer->buffer.mutex};
            auto head = std::exchange(transfer->buffer.buffer_head, nullptr);
            transfer->buffer.buffer_tail = &transfer->buffer.buffer_head;

            while(head) {
                auto next = head->next;
                free_buffer(head);
                head = next;
            }
            return;
        }

        Buffer* buffer{nullptr};
        size_t buffer_left_size{0};

        while(true) {
            {
                std::lock_guard block{transfer->buffer.mutex};
                buffer = transfer->buffer.buffer_head;
                buffer_left_size = transfer->buffer.bytes;
            }
            if(!buffer) {
                break;
            }

            const auto max_write_bytes = transfer->networking.throttle.bytes_left();
            if(!max_write_bytes) break; /* network throttle */

            assert(buffer->offset < buffer->length);
            auto written = ::send(fd, buffer->data + buffer->offset, std::min(buffer->length - buffer->offset, max_write_bytes), MSG_DONTWAIT | MSG_NOSIGNAL);
            if(written <= 0) {
                if(written == 0) {
                    /* EOF, how the hell is this event possible?! (Read should already catch it) */
                    logError(LOG_FT, "{} Client disconnected unexpectedly on write. Send {} bytes out of {}.",
                            transfer->log_prefix(), transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset);

                    transfer->handle->report_transfer_statistics(transfer->shared_from_this());
                    if(auto callback{transfer->handle->callback_transfer_aborted}; callback)
                        callback(transfer->transfer, { TransferError::UNEXPECTED_CLIENT_DISCONNECT, strerror(errno) });

                    {
                        std::unique_lock slock{transfer->state_mutex};
                        transfer->handle->disconnect_client(transfer->shared_from_this(), slock, true);
                    }
                } else {
                    if(errno == EAGAIN) {
                        transfer->add_network_write_event(false);
                        break;
                    }

                    logError(LOG_FT, "{} Received network write error. Send {} bytes out of {}. Closing transfer.",
                            transfer->log_prefix(), transfer->statistics.file_bytes_transferred, transfer->transfer->expected_file_size - transfer->transfer->file_offset);

                    transfer->handle->report_transfer_statistics(transfer->shared_from_this());
                    if(auto callback{transfer->handle->callback_transfer_aborted}; callback)
                        callback(transfer->transfer, { TransferError::NETWORK_IO_ERROR, strerror(errno) });

                    {
                        std::unique_lock slock{transfer->state_mutex};
                        transfer->handle->disconnect_client(transfer->shared_from_this(), slock, false);
                    }
                }
                return;
            } else {
                buffer->offset += written;
                assert(buffer->offset <= buffer->length);
                if(buffer->length == buffer->offset) {
                    {
                        std::lock_guard block{transfer->buffer.mutex};
                        transfer->buffer.buffer_head = buffer->next;
                        if(!buffer->next)
                            transfer->buffer.buffer_tail = &transfer->buffer.buffer_head;
                        assert(transfer->buffer.bytes >= written);
                        transfer->buffer.bytes -= written;
                        buffer_left_size = transfer->buffer.bytes;
                    }

                    free_buffer(buffer);
                } else {
                    std::lock_guard block{transfer->buffer.mutex};
                    assert(transfer->buffer.bytes >= written);
                    transfer->buffer.bytes -= written;
                    buffer_left_size = transfer->buffer.bytes;
                }

                transfer->timings.last_write = std::chrono::system_clock::now();
                transfer->statistics.network_bytes_send += written;

                if(transfer->networking.throttle.increase_bytes(written))
                    break; /* we've to slow down */
            }
        }

        if(buffer_left_size > 0)
            transfer->add_network_write_event(false);
        else if(transfer->state == FileClient::STATE_DISCONNECTING) {
            if(transfer->statistics.file_bytes_transferred + transfer->transfer->file_offset == transfer->transfer->expected_file_size) {
                logMessage(LOG_FT, "{} Finished file transfer within {}. Closing connection.", transfer->log_prefix(), duration_to_string(std::chrono::system_clock::now() - transfer->timings.key_received));
                transfer->handle->report_transfer_statistics(transfer->shared_from_this());
                if(auto callback{transfer->handle->callback_transfer_finished}; callback)
                    callback(transfer->transfer);
            } else {
                debugMessage(LOG_FT, "{} Flushed output buffer.", transfer->log_prefix());
            }
            std::unique_lock slock{transfer->state_mutex};
            transfer->handle->disconnect_client(transfer->shared_from_this(), slock, false);
            return;
        }
        transfer->handle->enqueue_disk_io(transfer->shared_from_this());
    }
}

inline std::string transfer_key_to_string(char key[TRANSFER_KEY_LENGTH]) {
    std::string result{};
    result.resize(TRANSFER_KEY_LENGTH);

    for(size_t index{0}; index < TRANSFER_KEY_LENGTH; index++)
        result[index] = isprint(key[index]) ? key[index] : '.';

    return result;
}

size_t LocalFileTransfer::handle_transfer_read(const std::shared_ptr<FileClient> &client, const char *buffer, size_t length) {
    if(client->state == FileClient::STATE_AWAITING_KEY) {
        if(client->networking.protocol == FileClient::PROTOCOL_UNKNOWN) {
            /* lets read the key bytes (16) and then decide */

            if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH) {
                const auto bytes_write = std::min(TRANSFER_KEY_LENGTH - client->transfer_key.provided_bytes, length);
                memcpy(client->transfer_key.key + client->transfer_key.provided_bytes, buffer, bytes_write);
                length -= bytes_write;
                buffer += bytes_write;
                client->transfer_key.provided_bytes += bytes_write;
            }

            if(client->transfer_key.provided_bytes < TRANSFER_KEY_LENGTH)
                return 0; /* we need more data */

            if(pipes::SSL::is_ssl((uint8_t*) client->transfer_key.key, client->transfer_key.provided_bytes)) {
                client->networking.protocol = FileClient::PROTOCOL_HTTPS;
                client->networking.max_read_buffer_size = (size_t) 1024 * 2; /* HTTPS-Header are sometimes a bit bigger. Dont cap max bandwidth here. */
                debugMessage(LOG_FT, "{} Using protocol HTTPS for file transfer.", client->log_prefix());

                char first_bytes[TRANSFER_KEY_LENGTH];
                memcpy(first_bytes, client->transfer_key.key, TRANSFER_KEY_LENGTH);
                client->transfer_key.provided_bytes = 0;

                this->handle_transfer_read(client, first_bytes, TRANSFER_KEY_LENGTH);
                return this->handle_transfer_read(client, buffer, length);
            } else {
                client->networking.protocol = FileClient::PROTOCOL_TS_V1;
                debugMessage(LOG_FT, "{} Using protocol RAWv1 for file transfer.", client->log_prefix());

                return this->handle_transfer_read(client, buffer, length);
            }
        } else if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) {
            auto key_result = this->handle_transfer_key_provided(client);
            switch(key_result) {
                case TransferKeyApplyResult::SUCCESS:
                    return length ? this->handle_transfer_read(client, buffer, length) : 0;

                case TransferKeyApplyResult::FILE_ERROR:
                    logMessage(LOG_FT, "{} Disconnecting client because we failed to open the target file.", client->log_prefix());
                    return (size_t) -1;

                case TransferKeyApplyResult::UNKNOWN_KEY:
                    logMessage(LOG_FT, "{} Disconnecting client because we don't recognise his key ({}).", client->log_prefix(), transfer_key_to_string(client->transfer_key.key));
                    return (size_t) -1;

                default:
                    logMessage(LOG_FT, "{} Disconnecting client because of an unknown key initialize error ({}).", client->log_prefix(), (int) key_result);
                    return (size_t) -1;
            }
        } else if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
            /* TODO! */
            /* 1. Await HTTP header
             * 2. Test for file transfer key & if its upload or download
             */
        } else {
            logError(LOG_FT, "{} Protocol variable contains unknown protocol. Disconnecting client.", client->log_prefix());

            std::unique_lock slock{client->state_mutex};
            client->handle->disconnect_client(client->shared_from_this(), slock, true);
            return (size_t) -1;
        }
    } else if(client->state == FileClient::STATE_TRANSFERRING) {
        assert(client->transfer);
        if(client->transfer->direction != Transfer::DIRECTION_UPLOAD) {
            debugMessage(LOG_FT, "{} Read {} bytes from client even though we're only sending a file. Ignoring it.", client->log_prefix(), length);
            return 0;
        }

        if(client->networking.protocol == FileClient::PROTOCOL_HTTPS) {
            client->networking.pipe_ssl.process_incoming_data(pipes::buffer_view{buffer, length});
            return client->buffer.bytes; /* a bit unexact but the best we could get away with it */
        } else if(client->networking.protocol == FileClient::PROTOCOL_TS_V1) {
            client->statistics.file_bytes_transferred += length;

            bool transfer_finished{false};
            auto writte_offset = client->statistics.file_bytes_transferred + client->transfer->file_offset;
            if(writte_offset > client->transfer->expected_file_size) {
                logMessage(LOG_FT, "{} Client send {} too many bytes (Transfer length was {}). Dropping them, flushing the disk IO and closing the transfer.", client->log_prefix(), writte_offset - client->transfer->expected_file_size, duration_to_string(std::chrono::system_clock::now() - client->timings.key_received));
                length -= writte_offset - client->transfer->expected_file_size;

                transfer_finished = true;
            } else if(writte_offset == client->transfer->expected_file_size) {
                logMessage(LOG_FT, "{} File upload has been completed in {}. Flushing disk IO and closing the connection.", client->log_prefix(), duration_to_string(std::chrono::system_clock::now() - client->timings.key_received));
                transfer_finished = true;
            }

            auto tbuffer = allocate_buffer(length);
            tbuffer->offset = 0;
            tbuffer->length = length;
            memcpy(tbuffer->data, buffer, length);

            size_t buffered_bytes;
            {
                std::lock_guard block{client->buffer.mutex};
                *client->buffer.buffer_tail = tbuffer;
                client->buffer.buffer_tail = &tbuffer->next;
                buffered_bytes = (client->buffer.bytes += length);
            }

            this->enqueue_disk_io(client);

            if(transfer_finished) {
                this->report_transfer_statistics(client);
                if(auto callback{this->callback_transfer_finished}; callback)
                    callback(client->transfer);

                std::unique_lock slock{client->state_mutex};
                client->handle->disconnect_client(client->shared_from_this(), slock, true);
            }
            return buffered_bytes;
        } else {
            logWarning(LOG_FT, "{} Read message for client with unknown protocol. Dropping {} bytes.", client->log_prefix(), length);
            return 0;
        }
    } else {
        logWarning(LOG_FT, "{} Read message at invalid client state ({}). Dropping message.", client->log_prefix(), client->state);
    }
    return 0;
}

TransferKeyApplyResult LocalFileTransfer::handle_transfer_key_provided(const std::shared_ptr<FileClient> &client) {
    {
        std::lock_guard tlock{this->transfers_mutex};
        for(auto it = this->pending_transfers.begin(); it != this->pending_transfers.end(); it++) {
            if(memcmp((*it)->transfer_key, client->transfer_key.key, TRANSFER_KEY_LENGTH) == 0) {
                client->transfer = *it;
                this->pending_transfers.erase(it);
                break;
            }
        }
    }

    if(!client->transfer) {
        logMessage(LOG_FT, "{} Received invalid transfer key. Disconnecting client.", client->log_prefix());

        std::unique_lock slock{client->state_mutex};
        client->handle->disconnect_client(client->shared_from_this(), slock, false);

        return TransferKeyApplyResult::UNKNOWN_KEY;
    }

    {
        std::string absolute_path{};
        auto transfer = client->transfer;
        switch (transfer->target_type) {
            case Transfer::TARGET_TYPE_AVATAR:
                absolute_path = this->file_system_.absolute_avatar_path(transfer->server_id, transfer->target_file_path);
                logMessage(LOG_FT, "{} Initialized avatar transfer for avatar \"{}\" ({} bytes, transferring {} bytes).",
                        client->log_prefix(), transfer->target_file_path, transfer->expected_file_size, transfer->expected_file_size - transfer->file_offset);
                break;
            case Transfer::TARGET_TYPE_ICON:
                absolute_path = this->file_system_.absolute_icon_path(transfer->server_id, transfer->target_file_path);
                logMessage(LOG_FT, "{} Initialized icon transfer for icon \"{}\" ({} bytes, transferring {} bytes).",
                        client->log_prefix(), transfer->target_file_path, transfer->expected_file_size, transfer->expected_file_size - transfer->file_offset);
                break;
            case Transfer::TARGET_TYPE_CHANNEL_FILE:
                absolute_path = this->file_system_.absolute_channel_path(transfer->server_id, transfer->channel_id, transfer->target_file_path);
                logMessage(LOG_FT, "{} Initialized channel transfer for file \"{}/{}\" ({} bytes, transferring {} bytes).",
                        client->log_prefix(), transfer->channel_id, transfer->target_file_path, transfer->expected_file_size, transfer->expected_file_size - transfer->file_offset);
                break;
            default:
                logError(LOG_FT, "{} Tried to initialize client with unknown file target type ({}). Dropping transfer.", client->log_prefix(), (int) transfer->target_type);
                this->report_transfer_statistics(client);
                if(auto callback{this->callback_transfer_aborted}; callback)
                    callback(client->transfer, { TransferError::UNKNOWN, "invalid transfer type" });

                {
                    std::unique_lock slock{client->state_mutex};
                    client->handle->disconnect_client(client->shared_from_this(), slock, false);
                }
                return TransferKeyApplyResult::FILE_ERROR;
        }
        debugMessage(LOG_FT, "{} Absolute file path: {}", client->log_prefix(), absolute_path);
        client->file.absolute_path = absolute_path;
    }

    if(client->transfer->max_bandwidth > 0) {
        debugMessage(LOG_FT, "{} Limit network bandwidth to {} bytes/second", client->log_prefix(), client->transfer->max_bandwidth);
        client->networking.throttle.set_max_bandwidth(client->transfer->max_bandwidth);
    }
    client->networking.max_read_buffer_size = (size_t) -1; /* limit max bandwidth via throttle */

    auto io_init_result = this->initialize_file_io(client);
    if(io_init_result != FileInitializeResult::SUCCESS) {
        logMessage(LOG_FT, "{} Failed to initialize file {}: {}/{}. Disconnecting client.",
                client->log_prefix(), client->transfer->direction == Transfer::DIRECTION_UPLOAD ? "writer" : "reader", (int) io_init_result, kFileInitializeResultMessages[(int) io_init_result]);

        this->report_transfer_statistics(client);
        if(auto callback{this->callback_transfer_aborted}; callback)
            callback(client->transfer, { TransferError::DISK_INITIALIZE_ERROR, std::to_string((int) io_init_result) + "/" + std::string{kFileInitializeResultMessages[(int) io_init_result]} });

        std::unique_lock slock{client->state_mutex};
        client->handle->disconnect_client(client->shared_from_this(), slock, false);

        return TransferKeyApplyResult::FILE_ERROR;
    }

    {
        std::unique_lock slock{client->state_mutex};
        if(client->state != FileClient::STATE_AWAITING_KEY)
            return TransferKeyApplyResult::SUCCESS; /* something disconnected the client */

        client->state = FileClient::STATE_TRANSFERRING;
    }

    if(auto callback{this->callback_transfer_started}; callback)
        callback(client->transfer);

    client->timings.key_received = std::chrono::system_clock::now();

    if(client->transfer->direction == Transfer::DIRECTION_DOWNLOAD)
        this->enqueue_disk_io(client); /* we've to take initiative */

    return TransferKeyApplyResult::SUCCESS;
}