
/*---------------------------------------------------------------------------*/
/* apirelay.cpp                                                              */
/* copyright (c) innovaphone 2016                                            */
/*                                                                           */
/*---------------------------------------------------------------------------*/

#include "platform/platform.h"
#include "common/os/iomux.h"
#include "common/interface/guuid.h"
#include "common/interface/socket.h"
#include "common/interface/webserver_plugin.h"
#include "common/interface/json_api.h"
#include "common/ilib/str.h"
#include "common/ilib/json.h"
#include "common/lib/appservice.h"
#include "common/lib/appwebsocket.h"
#include "apirelay.h"

/*---------------------------------------------------------------------------*/
/* class ApiRelayService                                                     */
/*---------------------------------------------------------------------------*/

ApiRelayService::ApiRelayService(class IIoMux * const iomux, ISocketProvider * localSocketProvider, class IWebserverPluginProvider * const webserverPluginProvider, AppServiceArgs * args, const char * localSocketPath) :
    AppService(iomux, localSocketProvider, args)
{
    this->iomux = iomux;
    this->webserverPluginProvider = webserverPluginProvider;
    this->localSocketProvider = localSocketProvider;
    this->localSocketPath = _strdup(localSocketPath);
}

ApiRelayService::~ApiRelayService()
{
    if (localSocketPath) free(localSocketPath);
}

class AppInstance * ApiRelayService::CreateInstance(AppInstanceArgs * args)
{
    return new ApiRelay(iomux, this, args);
}

/*---------------------------------------------------------------------------*/
/* class ApiRelay                                                            */
/*---------------------------------------------------------------------------*/

ApiRelay::ApiRelay(IIoMux * const iomux, class ApiRelayService * service, AppInstanceArgs * args) : AppInstance(service, args)
{
    this->iomux = iomux;
    this->service = service;
    this->webserverPlugin = service->webserverPluginProvider->CreateWebserverPlugin(iomux, service->localSocketProvider, this, args->webserver, args->webserverPath, this);
    this->stopping = false;
    this->activeClient = NULL;
    webserverPlugin->WebsocketListen();
    if (this->service->localSocketPath) {
        this->localSocket = this->service->localSocketProvider->CreateSocket(this->iomux, this, this, false);
        this->localSocket->Bind(this->service->localSocketPath);
        this->localSocket->Listen();
    }
    else {
        this->localSocket = NULL;
    }
}

ApiRelay::~ApiRelay()
{
}

const char * ApiRelay::GetAppPassword()
{
    return args.appPassword;
}

void ApiRelay::Stop()
{
    TryStop();
}

void ApiRelay::AppInstanceShutdownTimeout()
{
    debug->printf("ApiRelay::AppInstanceShutdownTimeout %d %d %d %d",
        !localSessions.empty(), !clientSessions.empty(), (webserverPlugin != NULL), (localSocket != NULL));
}

void ApiRelay::TryStop()
{
    if (!stopping) {
        stopping = true;
        if (!localSessions.empty()) {
            for (LocalSessions::iterator sessionItr = localSessions.begin(); sessionItr != localSessions.end(); sessionItr++) {
                (*sessionItr)->Close();
            }
        }
        if (!clientSessions.empty()) {
            for (ClientSessions::iterator sessionItr = clientSessions.begin(); sessionItr != clientSessions.end(); sessionItr++) {
                (*sessionItr)->Close();
            }
        }
        if (webserverPlugin) {
            webserverPlugin->Close();
        }
        if (localSocket) {
            localSocket->Shutdown();
        }
    }
    if (localSessions.empty() && clientSessions.empty() && !webserverPlugin && !localSocket) {
        appService->AppStopped(this);
    }
}

void ApiRelay::WebserverPluginWebsocketListenResult(IWebserverPlugin * plugin, const char * path, const char * registeredPathForRequest, const char * host)
{
    if (stopping) {
        plugin->Cancel(WSP_CANCEL_UNAVAILABLE);
        return;
    }
    if (!strcmp(path, "/client")) clientSessions.push_front(new ClientSession(this, plugin));
    else if (!strcmp(path, "/local")) localSessions.push_front(new LocalSession(this, plugin));
    else plugin->Cancel(WSP_CANCEL_NOT_FOUND);
}

void ApiRelay::WebserverPluginClose(IWebserverPlugin * plugin, wsp_close_reason_t reason, bool lastUser)
{
    if (lastUser) {
        delete this->webserverPlugin;
        this->webserverPlugin = 0;
        TryStop();
    }
}

void ApiRelay::SessionClosed(class ClientSession * session)
{
    for (LocalSessions::iterator sessionItr = localSessions.begin(); sessionItr != localSessions.end(); sessionItr++) {
        char sb[8192] = "\0";
        class json_io send(sb);
        word w = send.add_object(JSON_ID_ROOT, 0);
        send.add_string(w, "mt", "ApiConsumerClosed");
        send.add_string(w, "client", session->name);
        (*sessionItr)->AppWebsocketMessageSend(send, sb);
    }
    bool wasActiveClient = ActiveClient() == session;
    if (wasActiveClient) {
        if (ActiveClient()->apiObsolete) {
            for (LocalSessions::iterator sessionItr = localSessions.begin(); sessionItr != localSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(ActiveClient()->apiObsolete);
            }
        }
        activeClient = NULL;
    }
    clientSessions.remove(session);
    delete session;
    if (stopping) {
        TryStop();
    }
    else if (wasActiveClient && !clientSessions.empty()) {
        SetActiveClient(clientSessions.front());
    }
}

void ApiRelay::SessionClosed(class LocalSession * session)
{
    localSessions.remove(session);
    delete session;
    if (stopping) TryStop();
}

void ApiRelay::SetActiveClient(class ClientSession * session)
{
    if (session != ActiveClient()) {
        if (ActiveClient() && ActiveClient()->apiObsolete) {
            for (LocalSessions::iterator sessionItr = localSessions.begin(); sessionItr != localSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(ActiveClient()->apiObsolete);
            }
        }
        activeClient = session;
        for (LocalSessions::iterator sessionItr = localSessions.begin(); sessionItr != localSessions.end(); sessionItr++) {
            (*sessionItr)->AppWebsocketMessageSendText(GetApiUpdate());
        }
    }
}

static const char * EMPTY_API_UPDATE = "{\"mt\":\"ApiUpdate\",\"apis\":{}}";

const char * ApiRelay::GetApiUpdate()
{
    const char * result = 0;
    if (ActiveClient()) result = ActiveClient()->apiUpdate;
    if (!result) result = EMPTY_API_UPDATE;
    return result;
}

void ApiRelay::SocketListenResult(ISocket * const socket, const char * remoteAddr, word remotePort)
{
    if (stopping) {
        socket->Accept(NULL);
        return;
    }
    LocalSession * localSession = new LocalSession(this, NULL);
    localSessions.push_front(localSession);
    socket->Accept(localSession);
}

void ApiRelay::SocketShutdown(ISocket * const socket, shutdownreason_t reason)
{
    delete socket;
    localSocket = NULL;
    if (stopping) TryStop();
}

/*---------------------------------------------------------------------------*/
/* class ClientSession                                                       */
/*---------------------------------------------------------------------------*/

ClientSession::ClientSession(class ApiRelay * instance, class IWebserverPlugin * webserverPlugin) : AppWebsocket(webserverPlugin, instance)
{
    this->instance = instance;
    this->closing = false;
    this->appWebsocketClosed = false;
    this->connected = false;
    this->apiUpdate = 0;
    this->apiObsolete = 0;

    char tmp[256];
    char guid[128];
    IGuuid::GenerateUUID(guid);
    _sprintf(tmp, "@client-%s", guid);
    this->name = _strdup(tmp);
}

ClientSession::~ClientSession()
{
    if (apiUpdate) free(apiUpdate);
    if (apiObsolete) free(apiObsolete);
    if (name) free(name);
}

char * ClientSession::AppWebsocketPassword()
{
    return (char *)instance->GetAppPassword();
}

bool ClientSession::AppWebsocketConnectComplete(class json_io & msg, word base)
{
    connected = true;
    for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
        class ProvidedApi * api = 0;
        while ((api = (*sessionItr)->providedApis.GetNext(api))) {
            if (api->provideApi) {
                AppWebsocketMessageSendText(api->provideApi);
                if (api->apiModel) AppWebsocketMessageSendText(api->apiModel);
            }
        }
    }
    return true;
}

void ClientSession::AppWebsocketMessage(class json_io & msg, word base, const char * mt, const char * src)
{
    instance->SetActiveClient(this);
    if (!strcmp(mt, "ApiUpdate")) {
        char buffer[WS_MAX_DATA_SIZE];
        msg.add_replace_string(base, "mt", "ApiObsolete");
        msg.encode(base, buffer);
        if (apiObsolete) {
            for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(apiObsolete);
            }
            free(apiObsolete);
        }
        apiObsolete = _strdup(buffer);
        msg.add_replace_string(base, "mt", "ApiUpdate");
        msg.encode(base, buffer);
        if (apiUpdate) free(apiUpdate);
        apiUpdate = _strdup(buffer);
        for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
            (*sessionItr)->AppWebsocketMessageSendText(buffer);
        }
    }
    else if (!strcmp(mt, "ApiRequest")) {
        const char * api = msg.get_string(base, "apiId");
        const char * provider = msg.get_string(base, "provider");
        if (api && api[0] && provider && provider[0]) {
            msg.add_replace_string(base, "client", name);
            char buffer[WS_MAX_DATA_SIZE];
            msg.encode(base, buffer);
            for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
                if ((*sessionItr)->providedApis.Contains(api) && (!strcmp(provider, "*") || (!strcmp(provider, (*sessionItr)->name)))) {
                    (*sessionItr)->AppWebsocketMessageSendText(buffer);
                }
            }
        }
    }
    else if (!strcmp(mt, "ApiResult")) {
        const char * consumer = msg.get_string(base, "consumer");
        if (consumer && consumer[0]) {
            char buffer[WS_MAX_DATA_SIZE];
            msg.encode(base, buffer);
            for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
                if (!strcmp(consumer, (*sessionItr)->name)) {
                    (*sessionItr)->AppWebsocketMessageSendText(buffer);
                }
            }
        }
    }
    else if (!strcmp(mt, "ApiConsumerClosed")) {
        const char * api = msg.get_string(base, "apiId");
        const char * consumer = msg.get_string(base, "consumer");
        if (api && api[0] && consumer && consumer[0]) {
            msg.add_replace_string(base, "client", name);
            char buffer[WS_MAX_DATA_SIZE];
            msg.encode(base, buffer);
            for (LocalSessions::iterator sessionItr = instance->localSessions.begin(); sessionItr != instance->localSessions.end(); sessionItr++) {
                if ((*sessionItr)->providedApis.Contains(api)) {
                    (*sessionItr)->AppWebsocketMessageSendText(buffer);
                }
            }
        }
    }
    AppWebsocketMessageComplete();
}

void ClientSession::AppWebsocketClosed()
{
    appWebsocketClosed = true;
    TryClose();
}

void ClientSession::AppWebsocketMessageSend(class json_io & msg, char * buffer)
{
    if (connected) {
        AppWebsocket::AppWebsocketMessageSend(msg, buffer);
    }
}

void ClientSession::AppWebsocketMessageSendText(const char * buffer)
{
    if (connected) {
        AppWebsocket::AppWebsocketMessageSendText(buffer);
    }
}

void ClientSession::AppWebsocketMessageSendBinary(char * buffer, int len)
{
    if (connected) {
        AppWebsocket::AppWebsocketMessageSendBinary(buffer, len);
    }
}

void ClientSession::TryClose()
{
    closing = true;
    if (!appWebsocketClosed) {
        AppWebsocketClose();
        return;
    }
    instance->SessionClosed(this);
}

void ClientSession::Close()
{
    if (!closing) TryClose();
}

/*---------------------------------------------------------------------------*/
/* class LocalSession                                                        */
/*---------------------------------------------------------------------------*/

LocalSession::LocalSession(class ApiRelay * instance, class IWebserverPlugin * webserverPlugin) : AppWebsocket(webserverPlugin, instance)
{
    this->instance = instance;
    this->closing = false;
    this->appWebsocketClosed = false;
    this->connected = false;
    this->consumedApis = false;

    char tmp[256];
    char guid[128];
    IGuuid::GenerateUUID(guid);
    _sprintf(tmp, "@local-%s", guid);
    this->name = _strdup(tmp);
    this->localSocket = NULL;
    this->recvLen = 0;
    this->jsonLevel = 0;
    this->jsonQuotes = false;
    this->jsonEsc = false;
}

LocalSession::~LocalSession()
{
    if (name) free(name);
}

char * LocalSession::AppWebsocketPassword()
{
    return (char *)instance->GetAppPassword();
}

bool LocalSession::AppWebsocketConnectComplete(class json_io & msg, word base)
{
    connected = true;
    char sb[8192] = "\0";
    class json_io send(sb);
    word w = send.add_object(JSON_ID_ROOT, 0);
    send.add_string(w, "mt", "Info");
    send.add_string(w, "name", name);
    send.add_string(w, "title", dn);
    AppWebsocketMessageSend(send, sb);
    AppWebsocketMessageSendText(instance->GetApiUpdate());
    return true;
}

void LocalSession::AppWebsocketMessage(class json_io & msg, word base, const char * mt, const char * src)
{
    char buffer[WS_MAX_DATA_SIZE];
    if (!strcmp(mt, "RegisterApi")) {
        const char * api = msg.get_string(base, "apiId");
        msg.add_replace_string(base, "provider", name);
        msg.encode(base, buffer);
        providedApis.Add(api, buffer);
        for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
            (*sessionItr)->AppWebsocketMessageSendText(buffer);
        }
    }
    else if (!strcmp(mt, "DeregisterApi")) {
        const char * api = msg.get_string(base, "apiId");
        if (providedApis.Contains(api)) {
            providedApis.Remove(api);
            msg.add_replace_string(base, "provider", name);
            msg.encode(base, buffer);
            for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(buffer);
            }
        }
    }
    else if (!strcmp(mt, "ApiModel")) {
        const char * api = msg.get_string(base, "apiId");
        if (providedApis.Contains(api)) {
            msg.add_replace_string(base, "provider", name);
            msg.encode(base, buffer);
            providedApis.SetApiModel(api, buffer);
            for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(buffer);
            }
        }
    }
    else if (!strcmp(mt, "ApiRequest")) {
        consumedApis = true;
        msg.add_replace_string(base, "consumer", name);
        msg.encode(base, buffer);
        if (instance->ActiveClient()) instance->ActiveClient()->AppWebsocketMessageSendText(buffer);
    }
    else if (!strcmp(mt, "ApiResult")) {
        const char * client = msg.get_string(base, "client");
        if (client && client[0]) {
            msg.add_replace_string(base, "provider", name);
            msg.encode(base, buffer);
            for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
                if (!strcmp((*sessionItr)->name, client)) {
                    (*sessionItr)->AppWebsocketMessageSendText(buffer);
                }
            }
        }
    }
    if (!localSocket) {
        AppWebsocketMessageComplete();
    }
}

void LocalSession::AppWebsocketClosed()
{
    {
        class ProvidedApi * api = 0;
        while ((api = providedApis.GetNext(api))) {
            json_io msg(0);
            word base = msg.add_object(JSON_ID_ROOT, 0);
            msg.add_string(base, "mt", "DeregisterApi");
            msg.add_string(base, "apiId", api->name);
            msg.add_string(base, "provider", name);
            char buffer[WS_MAX_DATA_SIZE];
            msg.encode(base, buffer);
            for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
                (*sessionItr)->AppWebsocketMessageSendText(buffer);
            }
        }
    }

    if (consumedApis) {
        json_io msg(0);
        word base = msg.add_object(JSON_ID_ROOT, 0);
        msg.add_string(base, "mt", "ApiConsumerClosed");
        msg.add_string(base, "consumer", name);
        char buffer[WS_MAX_DATA_SIZE];
        msg.encode(base, buffer);
        for (ClientSessions::iterator sessionItr = instance->clientSessions.begin(); sessionItr != instance->clientSessions.end(); sessionItr++) {
            (*sessionItr)->AppWebsocketMessageSendText(buffer);
        }
    }

    appWebsocketClosed = true;
    TryClose();
}

void LocalSession::SocketAcceptComplete(ISocket * const socket)
{
    appWebsocketClosed = true;
    connected = true;
    localSocket = socket;
    char sb[8192] = "\0";
    class json_io send(sb);
    word w = send.add_object(JSON_ID_ROOT, 0);
    send.add_string(w, "mt", "Info");
    send.add_string(w, "name", name);
    send.add_string(w, "title", dn);
    AppWebsocketMessageSend(send, sb);
    AppWebsocketMessageSendText(instance->GetApiUpdate());
    socket->Recv(&recvBuf[recvLen], sizeof(recvBuf) - recvLen - 1, true);
}

void LocalSession::SocketRecvResult(ISocket * const socket, void * buf, size_t len)
{
    unsigned jsonStart;
    unsigned jsonPos;

    jsonStart = 0;
    jsonPos = recvLen;
    recvLen += len;
    while (jsonPos < recvLen) {
        while ((jsonPos < recvLen) && ((recvBuf[jsonPos] != '}') || (jsonLevel > 1) || jsonQuotes)) {
            if (jsonQuotes) {
                if (jsonEsc) {
                    jsonEsc = false;
                }
                else if (recvBuf[jsonPos] == '\\') {
                    jsonEsc = true;
                }
                else if (recvBuf[jsonPos] == '"') {
                    jsonQuotes = false;
                }
            }
            else if (recvBuf[jsonPos] == '"') {
                jsonQuotes = true;
            }
            else if (recvBuf[jsonPos] == '{') {
                jsonLevel++;
            }
            else if (recvBuf[jsonPos] == '}') {
                jsonLevel--;
            }
            jsonPos++;
        }
        if ((jsonPos < recvLen) && (recvBuf[jsonPos] == '}') && (jsonLevel == 1) && !jsonQuotes) {
            jsonPos++;
            recvBuf[jsonPos] = '\0';

            class json_io recv(&recvBuf[jsonStart]);
            if (recv.decode()) {
                word base = recv.get_object(JSON_ID_ROOT, NULL);
                AppWebsocketMessage(recv, base, recv.get_string(base, "mt"), recv.get_string(base, "src"));
            }

            recvBuf[jsonPos] = '{';
            jsonLevel = 0;
            jsonQuotes = false;
            jsonEsc = false;
            jsonStart = jsonPos;
        }
    }
    if (jsonStart != 0) {
        memmove(recvBuf, &recvBuf[jsonStart], recvLen - jsonStart);
        recvLen -= jsonStart;
    }
    socket->Recv(&recvBuf[recvLen], sizeof(recvBuf) - recvLen - 1, true);
}

void LocalSession::SocketShutdown(ISocket * const socket, shutdownreason_t reason)
{
    AppWebsocketClosed();
    delete socket;
    localSocket = NULL;
    TryClose();
}

void LocalSession::AppWebsocketMessageSend(class json_io & msg, char * buffer)
{
    if (connected) {
        if (localSocket) {
            localSocket->Send(buffer, msg.encode());
        }
        else {
            AppWebsocket::AppWebsocketMessageSend(msg, buffer);
        }
    }
}

void LocalSession::AppWebsocketMessageSendText(const char * buffer)
{
    if (connected) {
        if (localSocket) {
            localSocket->Send(buffer, strlen(buffer));
        }
        else {
            AppWebsocket::AppWebsocketMessageSendText(buffer);
        }
    }
}

void LocalSession::AppWebsocketMessageSendBinary(char * buffer, int len)
{
    if (connected) {
        if (localSocket) {
            localSocket->Send(buffer, len);
        }
        else {
            AppWebsocket::AppWebsocketMessageSendBinary(buffer, len);
        }
    }
}

void LocalSession::TryClose()
{
    closing = true;
    if (!appWebsocketClosed) {
        AppWebsocketClose();
        return;
    }
    if (localSocket) {
        localSocket->Shutdown();
        return;
    }
    instance->SessionClosed(this);
}

void LocalSession::Close()
{
    if (!closing) TryClose();
}

/*---------------------------------------------------------------------------*/
/* class ProvidedApiSet                                                      */
/*---------------------------------------------------------------------------*/

ProvidedApiSet::ProvidedApiSet()
{
    tree = 0;
}

ProvidedApiSet::~ProvidedApiSet()
{
    class ProvidedApi * out = 0;
    while ((out = (class ProvidedApi *)tree)) {
        tree = tree->btree_get(tree);
        delete out;
    }
}

void ProvidedApiSet::Add(const char * name, const char * message)
{
    if (name && !tree->btree_find(name)) tree = tree->btree_put(new class ProvidedApi(name, message));
}

void ProvidedApiSet::Remove(const char * name)
{
    class btree * out = tree->btree_find(name);
    if (out) {
        tree = tree->btree_get(out);
        delete out;
    }
}

bool ProvidedApiSet::Contains(const char * name)
{
    return tree->btree_find(name) != 0;
}

void ProvidedApiSet::SetProvideApi(const char * name, const char * message)
{
    class ProvidedApi * api = (class ProvidedApi *) tree->btree_find(name);
    if (api) api->SetProvideApi(message);
}

void ProvidedApiSet::SetApiModel(const char * name, const char * message)
{
    class ProvidedApi * api = (class ProvidedApi *) tree->btree_find(name);
    if (api) api->SetApiModel(message);
}

class ProvidedApi * ProvidedApiSet::GetNext(class ProvidedApi * api)
{
    return (class ProvidedApi *)tree->btree_find_next_right(api ? api->name : 0);
}

ProvidedApi::ProvidedApi(const char * name, const char * message)
{
    this->name = _strdup(name);
    this->provideApi = message ? _strdup(message) : 0;
    this->apiModel = 0;
}

ProvidedApi::~ProvidedApi()
{
    free(name);
    if (provideApi) free(provideApi);
    if (apiModel) free(apiModel);
}

void ProvidedApi::SetProvideApi(const char * message)
{
    if (this->provideApi) free(this->provideApi);
    this->provideApi = message ? _strdup(message) : 0;
}

void ProvidedApi::SetApiModel(const char * message)
{
    if (this->apiModel) free(this->apiModel);
    this->apiModel = message ? _strdup(message) : 0;
}

int ProvidedApi::btree_compare(void * key)
{
    return strcmp((const char *)key, name);
}

int ProvidedApi::btree_compare(class btree * b)
{
    return strcmp(((ProvidedApi *)b)->name, name);
}
