/*
 * .============.
 * // M A K E / \
 * // C++ DEV / \
 * // E A S Y / \/ \
 * ++ ----------. \/\ .
 * \\ \ \ /\ /
 * \\ \ \ /
 * \\ \ \ /
 * -============'
 *
 * Copyright (c) 2018 Hevake and contributors, all rights reserved.
 *
 * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
 * Use of this source code is governed by MIT license that can be found
 * in the LICENSE file in the root of the source tree. All contributing
 * project authors may be found in the CONTRIBUTORS.md file in the root
 * of the source tree.
 */
#include "server_imp.h"

#include <algorithm>

#include <tbox/base/log.h>
#include <tbox/base/assert.h>
#include <tbox/base/wrapped_recorder.h>
#include <tbox/util/buffer.h>
#include <tbox/network/tcp_connection.h>

#include "middleware.h"

namespace tbox {
namespace http {
namespace server {

using namespace std;
using namespace std::placeholders;
using namespace event;
using namespace network;

Server::Impl::Impl(Server *wp_parent, Loop *wp_loop)
 : wp_parent_(wp_parent)
 , wp_loop_(wp_loop)
 , tcp_server_(wp_loop)
{ }

Server::Impl::~Impl()
{
 TBOX_ASSERT(cb_level_ == 0);
 cleanup();
}

bool Server::Impl::initialize(const network::SockAddr &bind_addr, int listen_backlog)
{
 if (!tcp_server_.initialize(bind_addr, listen_backlog))
 return false;

 tcp_server_.setConnectedCallback(bind(&Impl::onTcpConnected, this, _1));
 tcp_server_.setReceiveCallback(bind(&Impl::onTcpReceived, this, _1, _2), 0);
 tcp_server_.setSendCompleteCallback(bind(&Impl::onTcpSendCompleted, this, _1));

 state_ = State::kInited;
 return true;
}

bool Server::Impl::setTlsConfig(const network::TlsConfig &config)
{
 return tcp_server_.setTlsConfig(config);
}

bool Server::Impl::start()
{
 if (tcp_server_.start()) {
 state_ = State::kRunning;
 return true;
 }
 return false;
}

void Server::Impl::stop()
{
 if (state_ == State::kRunning) {
 tcp_server_.stop();
 state_ = State::kInited;
 }
}

void Server::Impl::cleanup()
{
 if (state_ != State::kNone) {
 stop();

 mw_cabinet_.foreach([](RequestHandler *ptr) { delete ptr; });
 mw_cabinet_.clear();
 mw_order_.clear();
 tcp_server_.cleanup();

 state_ = State::kNone;
 }
}

MiddlewareToken Server::Impl::use(RequestHandler &&handler)
{
 if (cb_level_ > 0) {
 LogWarn("不能在 next 链中调用 use(),请使用 Loop 的 runNext() 来处理");
 return MiddlewareToken();
 }

 auto token = mw_cabinet_.alloc(new RequestHandler(std::move(handler)));
 mw_order_.push_back(token);
 return token;
}

MiddlewareToken Server::Impl::use(Middleware *wp_middleware)
{
 if (cb_level_ > 0) {
 LogWarn("不能在 next 链中调用 use(),请使用 Loop 的 runNext() 来处理");
 return MiddlewareToken();
 }

 auto token = mw_cabinet_.alloc(new RequestHandler(bind(&Middleware::handle, wp_middleware, _1, _2)));
 mw_order_.push_back(token);
 return token;
}

bool Server::Impl::unuse(const MiddlewareToken &token)
{
 if (cb_level_ > 0) {
 LogWarn("不能在 next 链中调用 unuse(),请使用 Loop 的 runNext() 来处理");
 return false;
 }

 //! 从 Cabinet 中释放
 RequestHandler *handler = mw_cabinet_.free(token);
 if (handler == nullptr)
 return false; //! token 无效或已释放
 delete handler;

 //! 从 order 中删除该 token(低频 O(n) 操作)
 auto iter = find(mw_order_.begin(), mw_order_.end(), token);
 if (iter != mw_order_.end())
 mw_order_.erase(iter);

 return true;
}

void Server::Impl::onTcpConnected(const TcpServer::ConnToken &ct)
{
 RECORD_SCOPE();
 auto conn = new Connection;
 tcp_server_.setContext(ct, conn,
 [](void* ptr) {
 auto conn = static_cast<Connection*>(ptr);
 CHECK_DELETE_OBJ(conn);
 }
 );
}

namespace {
bool IsLastRequest(const Request *req)
{
 auto iter = req->headers.find("Connection");
 if (req->http_ver == HttpVer::k1_0) {
 //! 1.0 版本默认为一连接一请求,需要特定的 Connection: Keep-Alive 才能持处
 if (iter == req->headers.end()) {
 return true;
 } else {
 return iter->second.find("keep-alive") == std::string::npos;
 }
 } else {
 //! 否则为 1.1 及以上的版本,默认为持久连接;除非出现 Connection: close
 if (iter == req->headers.end()) {
 return false;
 } else {
 return iter->second.find("close") != std::string::npos;
 }
 }
}
}

void Server::Impl::onTcpReceived(const TcpServer::ConnToken &ct, Buffer &buff)
{
 RECORD_SCOPE();
 Connection *conn = static_cast<Connection*>(tcp_server_.getContext(ct));
 TBOX_ASSERT(conn != nullptr);

 //! 如果已被标记为最后的请求,就不应该再有请求来
 if (conn->close_index != numeric_limits<int>::max()) {
 buff.hasReadAll();
 LogWarn("should not recv any data");
 return;
 }

 //! 如果已被标记为升级请求,停止解析 HTTP 数据
 if (conn->is_upgrade) {
 //! 不消费剩余数据,留给升级后的协议处理
 return;
 }

 while (buff.readableSize() > 0) {
 size_t rsize = conn->req_parser.parse(buff.readableBegin(), buff.readableSize());
 buff.hasRead(rsize);

 if (conn->req_parser.state() == RequestParser::State::kFinishedAll) {
 Request *req = conn->req_parser.getRequest();

 if (context_log_enable_)
 LogDbg("REQ: [%s]", req->toString().c_str());

 auto sp_ctx = make_shared<Context>(wp_parent_, ct, conn->req_index++, req);
 handle(sp_ctx, 0);

 //! 检查是否有协议升级回调(WebSocket、SSE 等)
 //! 如果有,标记连接为升级模式,停止继续解析 HTTP 数据
 if (sp_ctx->res().upgrade_cb) {
 conn->is_upgrade = true;
 //! 升级请求:保留 buffer 中未消费的数据,留给升级后的协议
 break;
 }

 //! 非升级请求:检查是否为最后一个请求
 if (IsLastRequest(req)) {
 conn->close_index = conn->req_index;
 LogDbg("mark close at %d", conn->close_index);

 tcp_server_.shutdown(ct, SHUT_RD);
 }

 } else if (conn->req_parser.state() == RequestParser::State::kFail) {
 LogNotice("parse http from %s fail", tcp_server_.getClientAddress(ct).toString().c_str());
 tcp_server_.disconnect(ct);
 break;

 } else {
 break;
 }
 }
}

void Server::Impl::onTcpSendCompleted(const TcpServer::ConnToken &ct)
{
 RECORD_SCOPE();
 if (!tcp_server_.isClientValid(ct))
 return;

 Connection *conn = static_cast<Connection*>(tcp_server_.getContext(ct));
 TBOX_ASSERT(conn != nullptr);

 //! 如果最后一个已完成发送,则断开连接
 if (conn->res_index > conn->close_index)
 tcp_server_.disconnect(ct);
}

/**
 * 为了保证管道化连接中Respond与Request的顺序一致性,要做特殊处理。
 * 如果所提交的index不是当前需要回复的res_index,那么就先暂存起来,等前面的发送完成后再发送;
 * 如果是,则可以直接回复。然后再将暂存中的未发送的其它数据也一同发送。
 *
 * 对于协议升级请求(WebSocket 101、SSE 200 等),发送完响应后,
 * 将从HTTP服务器分离TcpConnection,并通过Respond::upgrade_cb回调交给升级协议。
 */
void Server::Impl::commitRespond(const TcpServer::ConnToken &ct, int index, Respond *res)
{
 RECORD_SCOPE();
 if (!tcp_server_.isClientValid(ct)) {
 delete res;
 return;
 }

 //! 处理协议升级请求(WebSocket 101、SSE 200 等)
 //! 触发条件:res->upgrade_cb 已设置(中间件负责设置)
 if (res->upgrade_cb) {
 //! 注意:必须在 std::move(upgrade_cb) 之前调用 toString()
 //! 否则 move 后 upgrade_cb 为空,toString() 会误判为普通响应而添加 Content-Length
 //! SSE 响应添加 Content-Length:0 会导致浏览器认为响应已完成并断开重连
 const string content = res->toString();

 //! 发送响应后,将 TcpConnection 从 HTTP 服务器分离,交给升级协议
 auto upgrade_cb = std::move(res->upgrade_cb);
 delete res;

 tcp_server_.send(ct, content.data(), content.size());
 if (context_log_enable_)
 LogDbg("RES: [%s]", content.c_str());

 //! 当前回调结束后立即执行 detach(使用 runNext,更高效)
 //! 因为 commitRespond() 是在 Loop 线程中执行的
 wp_loop_->runNext([this, ct, upgrade_cb] {
 TcpConnection *tcp_conn = tcp_server_.detachConnection(ct);
 if (tcp_conn != nullptr)
 upgrade_cb(tcp_conn);
 else
 LogWarn("tcp_conn == nullptr");
 }, "HttpUpgrade: detach connection");

 return;
 }

 Connection *conn = static_cast<Connection*>(tcp_server_.getContext(ct));
 TBOX_ASSERT(conn != nullptr);

 if (index == conn->res_index) {
 //! 将当前的数据直接发送出去
 {
 const string &content = res->toString();
 tcp_server_.send(ct, content.data(), content.size());
 delete res;
 if (context_log_enable_)
 LogDbg("RES: [%s]", content.c_str());
 }

 ++conn->res_index;

 //! 如果当前这个回复是最后一个,则不需要发送缓存中的数据
 if (conn->res_index > conn->close_index)
 return;

 //! 尝试发送 conn->res_buff 缓冲中的数据
 auto &res_buff = conn->res_buff;
 auto iter = res_buff.find(conn->res_index);

 while (iter != res_buff.end()) {
 Respond *res = iter->second;
 {
 const string &content = res->toString();
 tcp_server_.send(ct, content.data(), content.size());
 delete res;
 if (context_log_enable_)
 LogDbg("RES: [%s]", content.c_str());
 }

 res_buff.erase(iter);
 ++conn->res_index;

 //! 如果当前这个回复是最后一个,则不需要再发送缓存中的数据
 if (conn->res_index > conn->close_index)
 return;

 iter = res_buff.find(conn->res_index);
 }
 } else {
 //! 放入到 conn.res_buff 中暂存
 conn->res_buff[index] = res;
 }
}

void Server::Impl::handle(ContextSptr sp_ctx, size_t index)
{
 RECORD_SCOPE();
 if (index >= mw_order_.size())
 return;

 auto token = mw_order_.at(index);
 RequestHandler *handler_ptr = mw_cabinet_.at(token);

 if (handler_ptr && *handler_ptr) {
 ++cb_level_;
 (*handler_ptr)(sp_ctx, std::bind(&Impl::handle, this, sp_ctx, index + 1));
 --cb_level_;
 } else {
 //! handler 已被移除或为空,跳过,执行下一个
 handle(sp_ctx, index + 1);
 }
}

Server::Impl::Connection::~Connection()
{
 //! 一定要记得清除非发送的Respond
 for (auto & item : res_buff)
 delete item.second;
}

}
}
}