/*
 * Copyright (C) 2014-2019 Amlogic, Inc. All rights reserved.
 *
 * All information contained herein is Amlogic confidential.
 *
 * This software is provided to you pursuant to Software License Agreement
 * (SLA) with Amlogic Inc ("Amlogic"). This software may be used
 * only in accordance with the terms of this agreement.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification is strictly prohibited without prior written permission from
 * Amlogic.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#define LOG_TAG "SubSocketServer"

#include "SocketServer.h"

#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/poll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>

#include "../SubtitleServer.h"
#include "DataSource.h"
#include "EventsTracker.h"
#include "IpcBuffer.h"
#include "Segment.h"
#include "SubtitleLog.h"
#define USE_UN true
#if USE_UN
#define UN_BASE_PATH "/tmp/subtitlesocketserver.sock"
#endif

/**
    payload is:
        startFlag   : 4bytes
        sessionID   : 4Bytes (TBD)
        magic       : 4bytes (for double confirm)
        payload Type: defined in PayloadType_t
        payload size: 4Bytes (indicate the size of payload, size is total_send_bytes - 4*5)
                      exclude this and above 4 items.
        payload data: TO BE PARSED data
*/

/**
    payload is:
        startFlag   : 4bytes
        sessionID   : 4Bytes (TBD)
        magic       : 4bytes (for double confirm)
        payload Type: defined in PayloadType_t
        payload size: 4Bytes (indicate the size of payload, size is total_send_bytes - 4*5)
                      exclude this and above 4 items.
        payload data: TO BE PARSED data
*/

static inline void dump(const char *buf, int size) {
    char str[64] = {0};

    for (int i = 0; i < size; i++) {
        char chars[6] = {0};
        sprintf(chars, "%02x ", buf[i]);
        strcat(str, chars);
        if (i % 8 == 7) {
            SUBTITLE_LOGI("%s", str);
            str[0] = str[1] = 0;
        }
    }
    SUBTITLE_LOGI("%s", str);
}

static std::mutex _g_inst_mutex;

SubSocketServer::SubSocketServer() : mExitRequested(false) {
    SUBTITLE_LOGI_FUNC_IN();
    mEventsTracker = std::make_shared<EventsTracker>(SubSocketServer::handleEvent);
}

SubSocketServer::~SubSocketServer() {
    SUBTITLE_LOGI_FUNC_IN();

    if (mEventsTracker != nullptr) {
        mEventsTracker->requestExit();
    }

    mExitRequested = true;
    mDispatchThread.join();
}

SubSocketServer *SubSocketServer::mInstance = nullptr;

SubSocketServer *SubSocketServer::GetInstance() {
    SUBTITLE_LOGI_FUNC_IN();

    std::unique_lock<std::mutex> autolock(_g_inst_mutex);
    if (mInstance == nullptr) {
        mInstance = new SubSocketServer();
    }

    return mInstance;
}

int SubSocketServer::serve() {
    std::thread t = std::thread(&SubSocketServer::__threadLoop, this);
    t.detach();
    return 0;
}

void SubSocketServer::__threadLoop() {
    while (!mExitRequested && threadLoop());
    SUBTITLE_LOGI("EXIT!!!");
}

bool SubSocketServer::threadLoop() {
    SUBTITLE_LOGI_FUNC_IN();
    int flag = 1;
    int sockFd = socket(USE_UN ? AF_UNIX : AF_INET, SOCK_STREAM, 0);
    if (sockFd < 0) return true;

    if ((setsockopt(sockFd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag))) < 0) {
        SUBTITLE_LOGE("setsockopt failed.\n");
        close(sockFd);
        return true;
    }

#if USE_UN
    std::string unPath = std::string(UN_BASE_PATH);

    struct sockaddr_un un{};
    bzero(&un, sizeof(un));
    un.sun_family = AF_UNIX;
    strcpy(un.sun_path, unPath.c_str());

    // Check if the socket is active, unlink if it is not active
    if (access(unPath.c_str(), F_OK) == 0) {
        int testFd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);
        if (testFd >= 0) {
            if (connect(testFd, (struct sockaddr *)&un,
                        offsetof(struct sockaddr_un, sun_path) + unPath.size()) == 0) {
                SUBTITLE_LOGE("socket path %s is already in use by another process.",
                              unPath.c_str());
                close(testFd);
                close(sockFd);
                return false;
            }
            close(testFd);
        }
        unlink(unPath.c_str());
    }

    size_t sockLen = offsetof(struct sockaddr_un, sun_path) + unPath.size();

    if (::bind(sockFd, (struct sockaddr *)&un, (socklen_t)sockLen) == -1) {
        SUBTITLE_LOGE("bind as UN fail. error=%d, err:%s\n", errno, strerror(errno));
        close(sockFd);
        return false;
    }
#else
    struct sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

    if (::bind(sockFd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
        SUBTITLE_LOGE("bind as INET fail. error=%d, err:%s\n", errno, strerror(errno));
        close(sockFd);
        return false;
    }
#endif

    if (::listen(sockFd, QUEUE_SIZE) == -1) {
        SUBTITLE_LOGE("listen fail.error=%d, err:%s\n", errno, strerror(errno));
        close(sockFd);
        return true;
    }

    mIsServing = true;

    SUBTITLE_LOGI("[startServerThread] listen success.\n");
    while (!mExitRequested) {
        struct sockaddr_in client_addr;
        socklen_t length = sizeof(client_addr);
        int connFd = accept(sockFd, (struct sockaddr *)&client_addr, &length);
        if (connFd < 0) {
            close(sockFd);
            return true;
        }

        SUBTITLE_LOGI("New connection coming: fd=%d", connFd);
        auto *dataObj = new DataObj_t;
        dataObj->obj1 = this;
        mEventsTracker->addFd(connFd, EventsTracker::EVENT_INPUT | EventsTracker::EVENT_ERROR,
                              dataObj);
    }

    SUBTITLE_LOGI("closed.\n");
    close(sockFd);
    mEventsTracker->requestExit();
    mIsServing = false;
    return true;
}

int SubSocketServer::handleEvent(int fd, int events, void *data) {
    auto *dataObj = static_cast<DataObj_t *>(data);
    if (dataObj == nullptr) {
        SUBTITLE_LOGE("data DataObj_t == null.");
        close(fd);
        return EventsTracker::RET_REMOVE;
    }

    auto subSocketServer = static_cast<SubSocketServer *>(dataObj->obj1);
    if (subSocketServer == nullptr) {
        SUBTITLE_LOGE("data SubSocketServer == null.");
        if (dataObj->obj2) {
            delete dataObj->obj2;
        }
        dataObj->reset();

        close(fd);
        return EventsTracker::RET_REMOVE;
    }

    if (events & EventsTracker::EVENT_INPUT) {
        SUBTITLE_LOGI("fd= %d, events= EVENT_INPUT", fd);

        int ret = subSocketServer->processData(fd, dataObj);
        if (ret == EventsTracker::RET_REMOVE) {
            if (dataObj->obj2 != nullptr) {
                delete dataObj->obj2;
            }
            dataObj->reset();
            close(fd);
        }
        return ret;

    } else if (events & EventsTracker::EVENT_ERROR) {
        SUBTITLE_LOGE("Error occurred for fd %d, remove.", fd);
        if (dataObj->obj2 != nullptr) {
            delete dataObj->obj2;
        }

        dataObj->reset();
        close(fd);
        return EventsTracker::RET_REMOVE;
    }

    return EventsTracker::RET_CONTINUE;
}

int SubSocketServer::processData(int fd, DataObj_t *dataObj) {
    if (!dataObj) {
        SUBTITLE_LOGE("processData: dataObj is null");
        return EventsTracker::RET_REMOVE;
    }

    // 1. Precondition for data
    if (dataObj->obj2 == nullptr) {
        SUBTITLE_LOGI("create new handle for fd %d", fd);
        IpcBuffer *buf = new (std::nothrow) IpcBuffer(512 * 1024, "socket_buffer");
        if (!buf) {
            SUBTITLE_LOGE("new IpcBuffer failed");
            return EventsTracker::RET_REMOVE;
        }
        dataObj->obj2 = buf;
    } else {
        SUBTITLE_LOGD("handle already created for fd %d", fd);
    }

    auto ipcBuffer = static_cast<IpcBuffer *>(dataObj->obj2);

    if (ipcBuffer->_data_header == nullptr) {
        ipcBuffer->_data_header = new (std::nothrow) IpcPackageHeader{};
        if (!ipcBuffer->_data_header) {
            SUBTITLE_LOGE("new IpcPackageHeader failed");
            return EventsTracker::RET_REMOVE;
        }
    }

    char recvBuf[1024] = {0};
    int retLen = ::recv(fd, recvBuf, sizeof(recvBuf), 0);
    if (retLen <= 0) {
        int sessionId = ipcBuffer->_data_header->sessionId;
        SUBTITLE_LOGI("Client broken, fd=%d, sessionId=%d, remove. errno = %m", fd, sessionId);
        auto subSocketServer = static_cast<SubSocketServer *>(dataObj->obj1);
        if (subSocketServer) subSocketServer->onRemoteDead(sessionId < 0 ? 0 : sessionId);
        return EventsTracker::RET_REMOVE;
    }

    if (mClients.empty()) {
        SUBTITLE_LOGE("mClients is empty, do not process now.");
        return EventsTracker::RET_CONTINUE;
    }

    // 2. Process data
    ipcBuffer->write(recvBuf, retLen);
    if (ipcBuffer->readableCount() < sizeof(IpcPackageHeader)) {
        SUBTITLE_LOGE("No enough data for Header.");
        return EventsTracker::RET_CONTINUE;
    }

    // Check sync word.
    int sizeOfInt = sizeof(int);
    char buffer[sizeof(IpcPackageHeader)] = {0};

    ipcBuffer->peek(buffer, sizeOfInt);
    int syncWord = peekAsSocketWord(buffer);
    if (syncWord == START_FLAG) {
        if (ipcBuffer->readableCount() < sizeof(IpcPackageHeader)) {
            SUBTITLE_LOGE("ReadableCount too small for full header");
            return EventsTracker::RET_CONTINUE;
        }

        ipcBuffer->read(buffer, sizeOfInt);
        ipcBuffer->read(buffer + sizeOfInt, sizeof(IpcPackageHeader) - sizeOfInt);
        ipcBuffer->_data_header->syncWord = syncWord;
        ipcBuffer->_data_header->sessionId = peekAsSocketWord(buffer + sizeOfInt);
        ipcBuffer->_data_header->magicWord = peekAsSocketWord(buffer + sizeOfInt * 2);
        ipcBuffer->_data_header->dataSize = peekAsSocketWord(buffer + sizeOfInt * 3);
        ipcBuffer->_data_header->pkgType = peekAsSocketWord(buffer + sizeOfInt * 4);

        SUBTITLE_LOGD(
            "Find header:\n"
            "struct IpcPackageHeader {\n"
            "    int syncWord = %d;\n"
            "    int sessionId = %d;\n"
            "    int magicWord = %d;\n"
            "    int dataSize = %d;\n"
            "    int pkgType = %d;\n"
            "};",
            ipcBuffer->_data_header->syncWord, ipcBuffer->_data_header->sessionId,
            ipcBuffer->_data_header->magicWord, ipcBuffer->_data_header->dataSize,
            ipcBuffer->_data_header->pkgType);

        const int TYPE_XEDC = 0x58454443;  // Reverse of eTypeSubtitleExitServ
        if (eTypeSubtitleExitServ == ipcBuffer->_data_header->pkgType ||
            TYPE_XEDC == ipcBuffer->_data_header->pkgType) {
            SUBTITLE_LOGI("exit requested!");
            return EventsTracker::RET_REMOVE;
        }
    }

    if (ipcBuffer->readableCount() < ipcBuffer->_data_header->dataSize) {
        // Maybe data is coming.
        SUBTITLE_LOGI("Buffer readableCount(%d) < Needed data size(%d), waiting more data.",
                      ipcBuffer->readableCount(), ipcBuffer->_data_header->dataSize);
        return EventsTracker::RET_CONTINUE;
    }

    char *payloads = (char *)malloc(ipcBuffer->_data_header->dataSize + sizeOfInt);
    if (payloads == nullptr) {
        SUBTITLE_LOGE("Memory allocation failed for payloads.");
        return EventsTracker::RET_REMOVE;
    }
    memcpy(payloads, &ipcBuffer->_data_header->pkgType, sizeOfInt);  // fill package type
    ipcBuffer->read(payloads + sizeOfInt, ipcBuffer->_data_header->dataSize);
    {  // notify listener
        std::lock_guard<std::mutex> guard(mLock);
        DataListener *listener = mClients.front();
        if (listener != nullptr) {
            if (listener->onData(payloads, ipcBuffer->_data_header->dataSize + sizeOfInt) < 0) {
                // if returen -1, means client exit, this request clientConnected exit!
                free(payloads);
                return EventsTracker::RET_REMOVE;
            }
        }
    }
    free(payloads);
    ipcBuffer->reset();

    return EventsTracker::RET_CONTINUE;
}

bool SubSocketServer::registClient(DataListener *client) {
    std::lock_guard<std::mutex> guard(GetInstance()->mLock);
    GetInstance()->mClients.push_back(client);
    SUBTITLE_LOGI("registClient: %p size=%zu", client, GetInstance()->mClients.size());
    return true;
}

bool SubSocketServer::unregisterClient(DataListener *client) {
    // obviously, BUG here! impl later, support multi-client.
    // TODO: revise the whole mClient, if we want to support multi subtitle

    std::lock_guard<std::mutex> guard(GetInstance()->mLock);

    std::vector<DataListener *> &vecs = GetInstance()->mClients;
    if (vecs.size() > 0) {
        for (auto it = vecs.cbegin(); it != vecs.cend(); it++) {
            if ((*it) == client) {
                vecs.erase(it);
                break;
            }
        }

        // GetInstance()->mClients.pop_back();
        SUBTITLE_LOGI("unregisterClient: %p size=%zu", client, GetInstance()->mClients.size());
    }
    return true;
}

void SubSocketServer::onRemoteDead(int sessionId) {
    SUBTITLE_LOGI("onRemoteDead, sessionId= %d", sessionId);

    SubtitleServer *server = SubtitleServer::Instance();
    if (server && !server->isClosed(sessionId)) {
        if (server->close(sessionId) == Result::OK &&
            server->closeConnection(sessionId) == Result::OK) {
            SUBTITLE_LOGI("onRemoteDead, self close OK for id %d", sessionId);
        } else {
            SUBTITLE_LOGI("onRemoteDead, self close failed for id %d", sessionId);
        }
    } else {
        SUBTITLE_LOGI("onRemoteDead, already closed for id %d", sessionId);
    }
}