Teaspeak-Server/server/src/client/voice/VoiceClient.cpp

#include <algorithm>
#include <memory>
#include <tomcrypt.h>
#include <arpa/inet.h>
#include <ThreadPool/Timer.h>
#include <misc/endianness.h>
#include <misc/memtracker.h>
#include <log/LogUtils.h>
#include <ThreadPool/ThreadHelper.h>

#include "VoiceClient.h"
#include "src/InstanceHandler.h"
#include "src/manager/ActionLogger.h"

using namespace std;
using namespace std::chrono;
using namespace ts::server;
using namespace ts::protocol;

constexpr static auto kMaxWhisperClientNameLength{30};
constexpr static auto kWhisperClientUniqueIdLength{28}; /* base64 encoded SHA1 hash */

VoiceClient::VoiceClient(const std::shared_ptr<VoiceServer>& server, const sockaddr_storage* address) :
    SpeakingClient{server->server->sql, server->server},
    voice_server(server) {
    assert(address);
    memtrack::allocated<VoiceClient>(this);
    memcpy(&this->remote_address, address, sizeof(sockaddr_storage));

    debugMessage(this->server->getServerId(), " Creating VoiceClient instance at {}", (void*) this);
}

void VoiceClient::initialize() {
    auto ref_self = dynamic_pointer_cast<VoiceClient>(this->ref());
    this->server_command_queue_ = std::make_unique<ServerCommandQueue>(
            serverInstance->server_command_executor(),
            std::make_unique<VoiceClientCommandHandler>(ref_self)
    );

    this->properties()[property::CLIENT_TYPE] = ClientType::CLIENT_TEAMSPEAK;
    this->properties()[property::CLIENT_TYPE_EXACT] = ClientType::CLIENT_TEAMSPEAK;

    this->state = ConnectionState::INIT_HIGH;
    this->connection = new connection::VoiceClientConnection(this);
}

VoiceClient::~VoiceClient() {
    debugMessage(this->getServerId(), " Deleting VoiceClient instance at {}", (void*) this);

    this->state = ConnectionState::DISCONNECTED;
    delete this->connection;
    this->connection = nullptr;

    if(this->flushing_thread) {
        logCritical(this->getServerId(), "Deleting a VoiceClient which should still be hold within the flush thread!");
        this->flushing_thread->detach();
    }

    memtrack::freed<VoiceClient>(this);
}

void VoiceClient::sendCommand0(const std::string_view& cmd, bool low, std::unique_ptr<std::function<void(bool)>> listener) {
    this->connection->send_command(cmd, low, std::move(listener));

#ifdef PKT_LOG_CMD
    logTrace(this->getServerId(), "{}[Command][Server -> Client] Sending command {}. Command low: {}. Full command: {}", CLIENT_STR_LOG_PREFIX, cmd.substr(0, cmd.find(' ')), low, cmd);
#endif
}

void VoiceClient::tick_server(const std::chrono::system_clock::time_point &time) {
    SpeakingClient::tick_server(time);
    {
        ALARM_TIMER(A1, "VoiceClient::tick", milliseconds(3));
        if(this->state == ConnectionState::CONNECTED) {
            this->connection->ping_handler().tick(time);
            this->connection->packet_statistics().tick();
        } else if(this->state == ConnectionState::INIT_LOW || this->state == ConnectionState::INIT_HIGH) {
            auto last_command = this->connection->crypt_setup_handler().last_handled_command();
            if(last_command.time_since_epoch().count() != 0) {
                if(time - last_command > seconds(5)) {
                    debugMessage(this->getServerId(), "{} Got handshake timeout. {}. State: {} Time: {}", CLIENT_STR_LOG_PREFIX,
                                 this->getLoggingPeerIp() + ":" + to_string(this->getPeerPort()),
                                 this->state == ConnectionState::INIT_HIGH ? "INIT_HIGH" : "INIT_LOW",
                                 duration_cast<seconds>(time - last_command).count()
                    );
                    this->close_connection(system_clock::now() + seconds(1));
                }
            }
        }
    }
}

std::chrono::milliseconds VoiceClient::current_ping() {
    return this->connection->ping_handler().current_ping();
}

bool VoiceClient::disconnect(const std::string &reason) {
    return this->disconnect(VREASON_SERVER_KICK, reason, this->server->serverRoot, true);
}

bool VoiceClient::disconnect(ts::ViewReasonId reason_id, const std::string &reason, const std::shared_ptr<ts::server::ConnectedClient>& invoker, bool notify_viewer) {
    /*
     * We don't have to lock the disconnect lock here, because we're not really unregistering the client.
     * Its only for the clients own flavour and everything which the client receives after will be ignored :)
     */

    ConnectionState old_state{};
    {
        std::lock_guard state_lock{this->state_lock};
        if(this->state == ConnectionState::DISCONNECTING || this->state == ConnectionState::DISCONNECTED)
            return false; //Already disconnecting/disconnected

        old_state = this->state;
        this->state = ConnectionState::DISCONNECTING;
    }

    if(old_state == ConnectionState::CONNECTED) {
        /* Client has been successflly initialized; Send normal disconnect. */

        Command cmd("notifyclientleftview");
        cmd["reasonmsg"] = reason;
        cmd["reasonid"] = reason_id;
        cmd["clid"] = this->getClientId();
        cmd["cfid"] = this->currentChannel ? this->currentChannel->channelId() : 0; //Failed when cid = 0????
        cmd["ctid"] = 0;

        if (invoker) {
            cmd["invokerid"] = invoker->getClientId();
            cmd["invokername"] = invoker->getDisplayName();
            cmd["invokeruid"] = invoker->getUid();
        }

        auto old_channel = this->currentChannel;
        if(old_channel) {
            serverInstance->action_logger()->client_channel_logger.log_client_leave(this->getServerId(), this->ref(), old_channel->channelId(), old_channel->name());
        }

        if(notify_viewer && this->server) {
            unique_lock channel_lock(this->server->channel_tree_lock);
            this->server->client_move(this->ref(), nullptr, invoker, reason, reason_id, false, channel_lock);
        } else {
            threads::MutexLock lock(this->command_lock);
            auto server_channel = dynamic_pointer_cast<ServerChannel>(this->currentChannel);
            if(server_channel)
                server_channel->unregister_client(this->ref());
            this->currentChannel = nullptr;
        }

        auto weak_self = this->weak_ref();
        this->sendCommand0(cmd.build(), false, std::make_unique<std::function<void(bool)>>([weak_self](bool success) {
            auto self = weak_self.lock();
            if(!self) {
                return;
            }

            if(!success) {
                /* In theory we have no need to disconnect the client any more since a failed acknowledge would do the trick for us */
                debugMessage(self->getServerId(), "{} Failed to receive disconnect acknowledge!", CLIENT_STR_LOG_PREFIX_(self));
            } else {
                debugMessage(self->getServerId(), "{} Received disconnect acknowledge!", CLIENT_STR_LOG_PREFIX_(self));
            }

            self->close_connection(chrono::system_clock::time_point{}); /* we received the ack, we do not need to flush anything */
        }));
    } else {
        //TODO: Extra case for INIT_HIGH?
        this->close_connection(chrono::system_clock::now() + chrono::seconds{5});
    }
    return true;
}

bool VoiceClient::close_connection(const system_clock::time_point &timeout) {
    auto self_lock  = dynamic_pointer_cast<VoiceClient>(this->ref());
    assert(self_lock); //Should never happen!

    bool flush = timeout.time_since_epoch().count() > 0;
    {
        std::lock_guard state_lock{this->state_lock};

        if(this->state == ConnectionState::DISCONNECTED) {
            return false;
        } else if(this->state == ConnectionState::DISCONNECTING) {
            /* here is nothing to pay attention for */
        } else if(this->state == ConnectionState::DISCONNECTING_FLUSHING) {
            if(!flush) {
                this->state = ConnectionState::DISCONNECTED;
                return true; /* the flush thread will execute the final disconnect */
            } else {
                //TODO: May update the flush timeout if its less then the other one?
                return true;
            }
        }
        this->state = flush ? ConnectionState::DISCONNECTING_FLUSHING : ConnectionState::DISCONNECTED;
    }

    debugMessage(this->getServerId(), "{} Closing voice client connection. (Flush: {})", CLIENT_STR_LOG_PREFIX, flush);
    //TODO: Move this out into a thread pool?
    if(this->flushing_thread && this->flushing_thread->joinable()) {
        logCritical(LOG_GENERAL, "VoiceClient::close_connection reached flushing thread with an active old handle. Ignoring request.");
        return true;
    }

    this->flushing_thread = std::make_shared<std::thread>([this, self_lock, timeout, flush]{
        {
            /* Await that all commands have been processed. It does not make sense to unregister the client while command handling. */
            std::lock_guard cmd_lock{this->command_lock};
        }

        if(flush) {
            debugMessage(this->getServerId(), "{} Awaiting write prepare, write and acknowledge queue flushed", CLIENT_STR_LOG_PREFIX);
            while(this->state == DISCONNECTING_FLUSHING) {
                if(system_clock::now() > timeout){
                    auto write_queue_flushed = this->connection->wait_empty_write_and_prepare_queue(timeout);
                    auto acknowledge_received = connection->packet_encoder().acknowledge_manager().awaiting_acknowledge() == 0;

                    if(write_queue_flushed && acknowledge_received)
                        break;

                    debugMessage(this->getServerId(), "{} Failed to flush pending messages. Acknowledges pending: {} Buffers pending: {}", CLIENT_STR_LOG_PREFIX, acknowledge_received, write_queue_flushed);
                    break;
                }
                if(!this->connection->wait_empty_write_and_prepare_queue(timeout))
                    continue;

                if(connection->packet_encoder().acknowledge_manager().awaiting_acknowledge() > 0) {
                    usleep(5000);
                    continue;
                }
                debugMessage(this->getServerId(), "{} Write and acknowledge queue are flushed", CLIENT_STR_LOG_PREFIX);
                break;
            }
        }

        if(this->state > DISCONNECTING) /* it could happen that the client "reconnects" while flushing this shit */
            this->finalDisconnect();
    });

    threads::name(*this->flushing_thread, "Flush thread VC");
    return true;
}

void VoiceClient::finalDisconnect() {
    auto ownLock = dynamic_pointer_cast<VoiceClient>(this->ref());
    assert(ownLock);

    lock_guard disconnect_lock_final(this->finalDisconnectLock);
    if(this->final_disconnected) {
        logError(this->getServerId(), "Tried to final disconnect {}/{} twice", this->getLoggingPeerIp() + ":" + to_string(this->getPeerPort()), this->getDisplayName());
        return;
    }
    this->final_disconnected = true;
    this->state = ConnectionState::DISCONNECTED;

    threads::MutexLock command_lock(this->command_lock); //We should not progress any commands while disconnecting
    //Unload manager cache
    this->processLeave();
    {
        if(this->flushing_thread) {
            this->flushing_thread->detach(); //The thread itself should be already done or executing this method
        }
        this->flushing_thread.reset();
    }
    if(this->voice_server) {
        this->voice_server->unregisterConnection(ownLock);
    }
}

void VoiceClient::send_voice_packet(const pipes::buffer_view &voice_buffer, const SpeakingClient::VoicePacketFlags &flags) {
    protocol::PacketFlags packet_flags{(uint8_t) PacketFlag::None};
    packet_flags |= flags.encrypted ? PacketFlag::None : PacketFlag::Unencrypted;
    packet_flags |= flags.head ? PacketFlag::Compressed : PacketFlag::None;
    packet_flags |= flags.fragmented ? PacketFlag::Fragmented : PacketFlag::None;
    packet_flags |= flags.new_protocol ? PacketFlag::NewProtocol : PacketFlag::None;

    this->connection->send_packet(PacketType::VOICE, packet_flags, voice_buffer.data_ptr<void>(), voice_buffer.length());
}

void VoiceClient::send_voice(const std::shared_ptr<SpeakingClient> &source_client, uint16_t seq_no, uint8_t codec, const void *payload, size_t payload_length) {
    /* TODO: Somehow set the head (compressed) flag for beginning voice packets? */
    auto packet = protocol::allocate_outgoing_server_packet(payload_length + 5);
    packet->type_and_flags_ = protocol::PacketType::VOICE;

    auto packet_payload = (uint8_t*) packet->payload;
    *((uint16_t*) packet_payload + 0) = htons(seq_no);
    *((uint16_t*) packet_payload + 1) = htons(source_client->getClientId());
    packet_payload[4] = codec;

    if(payload) {
        memcpy(packet->payload + 5, payload, payload_length);
    } else {
        assert(payload_length == 0);
    }

    this->getConnection()->send_packet(packet);
}

void VoiceClient::send_voice_whisper(const std::shared_ptr<SpeakingClient> &source_client, uint16_t seq_no, uint8_t codec, const void *payload, size_t payload_length) {
    bool head{false};
    if(this->whisper_head_counter < 5) {
        head = true;
        this->whisper_head_counter++;
    }
    if(!payload) {
        this->whisper_head_counter = 0;
    }

    protocol::OutgoingServerPacket* packet;
    size_t payload_offset{0};

    if(head && this->getType() == ClientType::CLIENT_TEASPEAK) {
        auto uniqueId = source_client->getUid();
        auto nickname = source_client->getDisplayName();

        if(uniqueId.length() > kWhisperClientUniqueIdLength) {
            logCritical(LOG_GENERAL, "Clients unique id is longer than the expected max length of {}. Unique length: {}", kWhisperClientUniqueIdLength, uniqueId.length());
            return;
        }

        if(nickname.length() > kMaxWhisperClientNameLength) {
            logCritical(LOG_GENERAL, "Clients name is longer than the expected max length of {}. Name length: {}", kMaxWhisperClientNameLength, nickname.length());
            return;
        }

        packet = protocol::allocate_outgoing_server_packet(
                payload_length + 5 + kWhisperClientUniqueIdLength + 1 + nickname.length());
        packet->type_and_flags_ |= protocol::PacketFlag::Compressed;

        memset(packet->payload + payload_offset, 0, kWhisperClientUniqueIdLength);
        memcpy(packet->payload + payload_offset, uniqueId.data(), uniqueId.length());
        payload_offset += kWhisperClientUniqueIdLength;

        packet->payload[payload_offset++] = nickname.length();
        memcpy(packet->payload + payload_offset, nickname.data(), nickname.length());
        payload_offset += nickname.length();
    } else {
        packet = protocol::allocate_outgoing_server_packet(payload_length + 5);
    }
    packet->type_and_flags_ |= protocol::PacketType::VOICE_WHISPER;

    *((uint16_t*) &packet->payload[payload_offset]) = htons(seq_no);
    payload_offset += 2;

    *((uint16_t*) &packet->payload[payload_offset]) = htons(source_client->getClientId());
    payload_offset += 2;

    packet->payload[payload_offset] = codec;
    payload_offset += 1;

    if(payload) {
        memcpy(packet->payload + payload_offset, payload, payload_length);
        payload_offset += payload_length;
    } else {
        assert(payload_length == 0);
    }
    packet->payload_size = payload_offset;

    this->getConnection()->send_packet(packet);
}

float VoiceClient::current_ping_deviation() {
    return this->connection->packet_encoder().acknowledge_manager().current_rttvar();
}

float VoiceClient::current_packet_loss() const {
    return this->connection->packet_statistics().current_packet_loss();
}
void VoiceClient::processJoin() {
    SpeakingClient::processJoin();
    if(this->rtc_client_id > 0) {
        {
            /* Normal audio channel */
            auto sender = this->server->rtc_server().create_audio_source_supplier_sender(this->rtc_client_id, 1);
            assert(sender.has_value());
            this->rtc_audio_supplier.reset(*sender);
        }

        {
            /* Audio whisper channel */
            auto sender = this->server->rtc_server().create_audio_source_supplier_sender(this->rtc_client_id, 2);
            assert(sender.has_value());
            this->rtc_audio_whisper_supplier.reset(*sender);
        }
    }
}