/*
 * .============.
 * // M A K E / \
 * // C++ DEV / \
 * // E A S E / \/ \
 * ++ ----------. \/\ .
 * \\ \ \ /\ /
 * \\ \ \ /
 * \\ \ \ /
 * -============'
 *
 * Copyright (c) 2026 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 "client_impl.h"

#include <tbox/base/log.h>
#include <tbox/base/assert.h>
#include <tbox/base/wrapped_recorder.h>
#include <tbox/event/timer_event.h>
#include <tbox/util/buffer.h>

namespace tbox {
namespace http {
namespace client {

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

Client::Impl::Impl(Client *wp_parent, Loop *wp_loop)
 : wp_parent_(wp_parent)
 , wp_loop_(wp_loop)
 , tcp_client_(wp_loop)
{ }

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

bool Client::Impl::initialize(const SockAddr &server_addr)
{
 if (state_ != State::kNone) {
 LogWarn("state not right, can't initialize");
 return false;
 }

 if (!tcp_client_.initialize(server_addr))
 return false;

 tcp_client_.setConnectedCallback(bind(&Impl::onTcpConnected, this));
 tcp_client_.setDisconnectedCallback(bind(&Impl::onTcpDisconnected, this));
 tcp_client_.setReceiveCallback(bind(&Impl::onTcpReceived, this, _1), 0);

 state_ = State::kInited;
 return true;
}

void Client::Impl::setTlsConfig(const TlsConfig &config)
{
 tcp_client_.setTlsConfig(config);
}

bool Client::Impl::start()
{
 if (state_ != State::kInited) {
 LogWarn("state not right, can't start");
 return false;
 }

 if (tcp_client_.start()) {
 state_ = State::kConnecting;
 return true;
 }
 return false;
}

void Client::Impl::stop()
{
 if (state_ == State::kConnected || state_ == State::kConnecting || state_ == State::kReconnWaiting) {
 tcp_client_.stop();
 //! 清理所有缓存和待处理的请求
 failAllPendingRequests(StatusCode::k503_ServiceUnavailable, "client stopped");
 for (auto &item : cached_requests_)
 CHECK_DELETE_RESET_OBJ(item.sp_req);
 cached_requests_.clear();
 res_parser_.reset();
 state_ = State::kInited;
 }
}

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

 //! 释放所有超时定时器
 for (auto &item : pending_requests_)
 CHECK_DELETE_RESET_OBJ(item.sp_timer);
 pending_requests_.clear();

 //! 释放所有缓存请求
 for (auto &item : cached_requests_)
 CHECK_DELETE_RESET_OBJ(item.sp_req);
 cached_requests_.clear();

 tcp_client_.cleanup();

 state_ = State::kNone;
 }
}

void Client::Impl::request(const Request &req, const RespondCallback &cb)
{
 if (state_ == State::kNone) {
 LogWarn("state is kNone, need initialize first");
 return;
 }

 sendRequest(req, cb);
}

void Client::Impl::request(Method method, const string &path, const RespondCallback &cb)
{
 Request req;
 req.method = method;
 req.http_ver = HttpVer::k1_1;
 req.url.path = path;
 request(req, cb);
}

void Client::Impl::request(Method method, const string &path, const string &body,
 const Headers &headers, const RespondCallback &cb)
{
 Request req;
 req.method = method;
 req.http_ver = HttpVer::k1_1;
 req.url.path = path;
 req.body = body;
 req.headers = headers;
 request(req, cb);
}

void Client::Impl::setAutoReconnect(bool enable)
{
 auto_reconnect_ = enable;
 tcp_client_.setAutoReconnect(enable);
}

void Client::Impl::setReconnectDelayCalcFunc(const ReconnectDelayCalc &func)
{
 tcp_client_.setReconnectDelayCalcFunc(func);
}

void Client::Impl::setRequestTimeout(chrono::milliseconds ms)
{
 request_timeout_ms_ = ms;
}

void Client::Impl::setContextLogEnable(bool enable)
{
 context_log_enable_ = enable;
}

void Client::Impl::setConnectedCallback(const ConnectedCallback &cb)
{
 connected_cb_ = cb;
}

void Client::Impl::setConnectFailCallback(const ConnectFailCallback &cb)
{
 connect_fail_cb_ = cb;
}

void Client::Impl::setDisconnectedCallback(const DisconnectedCallback &cb)
{
 disconnected_cb_ = cb;
}

//! ============ TCP 连接回调 ============

void Client::Impl::onTcpConnected()
{
 RECORD_SCOPE();
 state_ = State::kConnected;

 LogDbg("connected");

 //! 发送缓存中的请求
 sendCachedRequests();

 if (connected_cb_) {
 ++cb_level_;
 connected_cb_();
 --cb_level_;
 }
}

void Client::Impl::onTcpConnectFail()
{
 RECORD_SCOPE();
 LogDbg("connect fail");

 //! failAllPendingRequests 中会处理所有 pending 请求
 //! 但 connect fail 时,pending_requests_ 应为空(因为未连接时请求在 cached_requests_ 中)
 //! cached_requests_ 中的请求也要失败回调
 for (auto &item : cached_requests_) {
 Respond res;
 res.http_ver = HttpVer::k1_1;
 res.status_code = StatusCode::k503_ServiceUnavailable;
 res.body = "connect fail";
 if (item.cb) {
 ++cb_level_;
 item.cb(res);
 --cb_level_;
 }
 CHECK_DELETE_RESET_OBJ(item.sp_req);
 }
 cached_requests_.clear();

 if (connect_fail_cb_) {
 ++cb_level_;
 connect_fail_cb_();
 --cb_level_;
 }

 if (!auto_reconnect_) {
 state_ = State::kInited;
 } else {
 state_ = State::kReconnWaiting;
 }
}

void Client::Impl::onTcpDisconnected()
{
 RECORD_SCOPE();
 LogDbg("disconnected");

 //! 对所有 pending 请求执行错误回调
 failAllPendingRequests(StatusCode::k504_GatewayTimeout, "connection lost");

 res_parser_.reset();

 if (disconnected_cb_) {
 ++cb_level_;
 disconnected_cb_();
 --cb_level_;
 }

 if (auto_reconnect_) {
 state_ = State::kReconnWaiting;
 } else {
 state_ = State::kInited;
 }
}

void Client::Impl::onTcpReceived(Buffer &buff)
{
 RECORD_SCOPE();

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

 if (res_parser_.state() == RespondParser::State::kFinishedAll) {
 Respond *sp_respond = res_parser_.getRespond();
 if (sp_respond == nullptr) {
 LogWarn("getRespond() return nullptr, skip");
 continue;
 }

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

 //! 取出 pending_requests_ 队首的请求
 if (!pending_requests_.empty()) {
 auto &pending = pending_requests_.front();

 //! 停止超时定时器
 if (pending.sp_timer != nullptr) {
 pending.sp_timer->disable();
 CHECK_DELETE_RESET_OBJ(pending.sp_timer);
 }

 //! 调用回调
 if (pending.cb) {
 ++cb_level_;
 pending.cb(*sp_respond);
 --cb_level_;
 }

 pending_requests_.pop_front();
 } else {
 LogWarn("no pending request for this respond");
 }

 delete sp_respond;

 } else if (res_parser_.state() == RespondParser::State::kFail) {
 LogNotice("parse http respond fail");
 failAllPendingRequests(StatusCode::k502_BadGateway, "respond parse fail");
 //! 断开连接,让 TcpClient 重连
 tcp_client_.stop();
 if (auto_reconnect_) {
 tcp_client_.start();
 state_ = State::kReconnWaiting;
 } else {
 state_ = State::kInited;
 }
 break;

 } else {
 //! 数据不完整,等待更多数据
 break;
 }
 }
}

//! ============ 请求发送 ============

void Client::Impl::sendRequest(const Request &req, const RespondCallback &cb)
{
 if (state_ == State::kConnected) {
 //! 已连接,直接发送
 const string &content = req.toString();
 tcp_client_.send(content.data(), content.size());

 if (context_log_enable_)
 LogDbg("REQ: [%s]", content.c_str());

 //! 创建 PendingRequest,加入队列
 PendingRequest pending;
 pending.req_id = next_req_id_++;
 pending.cb = cb;

 //! 创建超时定时器
 if (request_timeout_ms_.count() > 0) {
 auto sp_timer = wp_loop_->newTimerEvent();
 sp_timer->initialize(request_timeout_ms_, Event::Mode::kOneshot);
 sp_timer->setCallback(bind(&Impl::onRequestTimeout, this, pending.req_id));
 sp_timer->enable();
 pending.sp_timer = sp_timer;
 }

 pending_requests_.push_back(pending);

 } else {
 //! 未连接,缓存请求,等连接建立后发送
 CachedRequest cached;
 cached.sp_req = new Request(req);
 cached.cb = cb;
 cached_requests_.push_back(cached);
 }
}

void Client::Impl::sendCachedRequests()
{
 //! 将缓存中的所有请求发送出去
 for (auto &item : cached_requests_) {
 sendRequest(*item.sp_req, item.cb);
 CHECK_DELETE_RESET_OBJ(item.sp_req);
 }
 cached_requests_.clear();
}

//! ============ 超时与失败处理 ============

void Client::Impl::onRequestTimeout(int req_id)
{
 RECORD_SCOPE();
 LogDbg("request %d timeout", req_id);

 //! 在 pending_requests_ 中查找该请求
 for (auto iter = pending_requests_.begin(); iter != pending_requests_.end(); ++iter) {
 if (iter->req_id == req_id) {
 //! 构造超时响应
 Respond res;
 res.http_ver = HttpVer::k1_1;
 res.status_code = StatusCode::k408_RequestTimeout;
 res.body = "request timeout";

 if (iter->cb) {
 ++cb_level_;
 iter->cb(res);
 --cb_level_;
 }

 //! 清理定时器
 CHECK_DELETE_RESET_OBJ(iter->sp_timer);

 //! 从队列中移除
 //! 注意:超时中间的请求被移除后,后续请求仍可正常接收响应
 //! 但如果服务器按序回复,中间的请求超时意味着后续请求可能也无法收到响应
 //! 这里采用保守策略:仅移除超时的请求,不影响其他请求
 pending_requests_.erase(iter);
 break;
 }
 }
}

void Client::Impl::failAllPendingRequests(StatusCode status_code, const string &message)
{
 for (auto &item : pending_requests_) {
 //! 停止超时定时器
 if (item.sp_timer != nullptr) {
 item.sp_timer->disable();
 CHECK_DELETE_RESET_OBJ(item.sp_timer);
 }

 //! 构造错误响应
 Respond res;
 res.http_ver = HttpVer::k1_1;
 res.status_code = status_code;
 res.body = message;

 if (item.cb) {
 ++cb_level_;
 item.cb(res);
 --cb_level_;
 }
 }
 pending_requests_.clear();

 //! 也处理缓存中的请求
 for (auto &item : cached_requests_) {
 Respond res;
 res.http_ver = HttpVer::k1_1;
 res.status_code = status_code;
 res.body = message;

 if (item.cb) {
 ++cb_level_;
 item.cb(res);
 --cb_level_;
 }
 CHECK_DELETE_RESET_OBJ(item.sp_req);
 }
 cached_requests_.clear();
}

}
}
}