/**
 * C++ High-Performance NATS Stock Level 2 Market Data Subscriber Example.
 * Uses libnats-c and header-only spdlog.
 * Includes system time and formatted exchange time.
 *
 * Compile command example on remote server:
 * g++ -O3 -std=c++20 -I. nats_client.cpp -o nats_client -lnats -lspdlog -lfmt
 * -lpthread
 */

#define SPDLOG_HEADER_ONLY // Use header-only mode to avoid library version
                           // mismatches

#include <atomic>
#include <chrono>
#include <iostream>
#include <signal.h>
#include <string>
#include <thread>
#include <ctime>

#include "MarketDataPOD.h"
#include <nats/nats.h>

// spdlog headers
#include "spdlog/async.h"
#include "spdlog/sinks/stdout_color_sinks.h"
#include "spdlog/spdlog.h"

// Graceful exit flag
std::atomic<bool> g_running{true};

void signal_handler(int) { g_running = false; }

// Clean helper to extract symbol from 32-byte char array safely
inline std::string CleanSymbol(const char *symbol) {
  int len = 0;
  while (len < 32 && symbol[len] != '\0') {
    len++;
  }
  return std::string(symbol, len);
}

// Get high-precision system time string (HH:MM:SS.mmm)
inline std::string GetSystemTimeStr() {
  auto now = std::chrono::system_clock::now();
  auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()) % 1000;
  auto timer = std::chrono::system_clock::to_time_t(now);
  std::tm bt = *std::localtime(&timer);
  char buf[64];
  std::snprintf(buf, sizeof(buf), "%02d:%02d:%02d.%03d", 
                bt.tm_hour, bt.tm_min, bt.tm_sec, (int)ms.count());
  return std::string(buf);
}

// Format integer exchange time (HHMMSSmmm) into (HH:MM:SS.mmm)
inline std::string FormatExchangeTime(int32_t t) {
  int ms = t % 1000;
  t /= 1000;
  int s = t % 100;
  t /= 100;
  int m = t % 100;
  int h = t / 100;
  char buf[64];
  std::snprintf(buf, sizeof(buf), "%02d:%02d:%02d.%03d", h, m, s, ms);
  return std::string(buf);
}

// 1. Snapshot callback handler (stk.l2.>)
void onSnapshotMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg,
                   void *closure) {
  int len = natsMsg_GetDataLength(msg);
  if (len < (int)sizeof(StkL2Snapshot)) {
    natsMsg_Destroy(msg);
    return;
  }
  const StkL2Snapshot *snap = (const StkL2Snapshot *)natsMsg_GetData(msg);

  SPDLOG_INFO("[SNAPSHOT] Topic: {} | Symbol: {} | Last: {:.2f} | Volume: {} | "
              "Turnover: {} | ExchangeTime: {} | SysTime: {}",
              natsMsg_GetSubject(msg), CleanSymbol(snap->symbol),
              snap->last / 10000.0, snap->volume, snap->turnover,
              FormatExchangeTime(snap->time), GetSystemTimeStr());

  natsMsg_Destroy(msg);
}

// 2. Transaction callback handler (stk.trans.>)
void onTransactionMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg,
                      void *closure) {
  int len = natsMsg_GetDataLength(msg);
  if (len < (int)sizeof(StkTransaction)) {
    natsMsg_Destroy(msg);
    return;
  }
  const StkTransaction *tx = (const StkTransaction *)natsMsg_GetData(msg);

  SPDLOG_INFO("[TRANS] Topic: {} | Symbol: {} | Index: {} | Price: {:.2f} | "
              "Vol: {} | BS: {} | ExchangeTime: {} | SysTime: {}",
              natsMsg_GetSubject(msg), CleanSymbol(tx->symbol), tx->index,
              tx->price / 10000.0, tx->volume, tx->bs_flag,
              FormatExchangeTime(tx->time), GetSystemTimeStr());

  natsMsg_Destroy(msg);
}

// 3. Order callback handler (stk.order.>)
void onOrderMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg,
                void *closure) {
  int len = natsMsg_GetDataLength(msg);
  if (len < (int)sizeof(StkOrder)) {
    natsMsg_Destroy(msg);
    return;
  }
  const StkOrder *ord = (const StkOrder *)natsMsg_GetData(msg);

  SPDLOG_INFO("[ORDER] Topic: {} | Symbol: {} | Index: {} | OrderNo: {} | Price: {:.2f} | "
              "Vol: {} | Kind: {} | ExchangeTime: {} | SysTime: {}",
              natsMsg_GetSubject(msg), CleanSymbol(ord->symbol), ord->index,
              ord->order_no, ord->price / 10000.0, ord->volume, ord->order_kind,
              FormatExchangeTime(ord->time), GetSystemTimeStr());

  natsMsg_Destroy(msg);
}

// 4. Index callback handler (stk.index.>)
void onIndexMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg,
                void *closure) {
  int len = natsMsg_GetDataLength(msg);
  if (len < (int)sizeof(StkIndex)) {
    natsMsg_Destroy(msg);
    return;
  }
  const StkIndex *idx = (const StkIndex *)natsMsg_GetData(msg);

  SPDLOG_INFO("[INDEX] Topic: {} | Symbol: {} | Last: {:.2f} | Volume: {} | "
              "Turnover: {} | ExchangeTime: {} | SysTime: {}",
              natsMsg_GetSubject(msg), CleanSymbol(idx->symbol),
              idx->last / 10000.0, idx->volume, idx->turnover,
              FormatExchangeTime(idx->time), GetSystemTimeStr());

  natsMsg_Destroy(msg);
}


int main(int argc, char *argv[]) {
  // Register signal handlers
  signal(SIGINT, signal_handler);
  signal(SIGTERM, signal_handler);

  // 1. Initialize spdlog async logger for high performance
  spdlog::init_thread_pool(8192, 1);
  auto stdout_sink = std::make_shared<spdlog::sinks::stdout_color_sink_mt>();
  auto async_logger = std::make_shared<spdlog::async_logger>(
      "async_nats_logger", stdout_sink, spdlog::thread_pool(),
      spdlog::async_overflow_policy::block);
  spdlog::set_default_logger(async_logger);
  spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] %v");
  spdlog::set_level(spdlog::level::info);

  SPDLOG_INFO("=================================================");
  SPDLOG_INFO("    NATS C++ High-Performance Subscriber Case     ");
  SPDLOG_INFO("=================================================");

  const char *nats_url = "nats://quote5.base32.cn:4222";
  if (argc > 1) {
    nats_url = argv[1];
  }

  // 2. Setup NATS connection options & authentication
  natsStatus s;
  natsConnection *conn = nullptr;
  natsOptions *opts = nullptr;

  s = natsOptions_Create(&opts);
  if (s != NATS_OK) {
    SPDLOG_ERROR("Failed to create NATS options: {}", natsStatus_GetText(s));
    return 1;
  }

  // Configure connection, credentials, and reconnection policy
  natsOptions_SetURL(opts, nats_url);
  natsOptions_SetUserInfo(opts, "level2_test", "level2_test");
  natsOptions_SetAllowReconnect(opts, true);
  natsOptions_SetMaxReconnect(opts, -1);
  natsOptions_SetReconnectWait(opts, 2000);
  natsOptions_SetReconnectBufSize(opts, 64 * 1024 * 1024);
  natsOptions_SetSendAsap(opts, true);

  // 3. Connect to NATS
  SPDLOG_INFO("Connecting to NATS server at {} ...", nats_url);
  s = natsConnection_Connect(&conn, opts);
  natsOptions_Destroy(opts);

  if (s != NATS_OK) {
    SPDLOG_ERROR("NATS connection failed: {}", natsStatus_GetText(s));
    return 1;
  }
  SPDLOG_INFO("Connected successfully to NATS!");

  // 4. Register Asynchronous Subscriptions with Callback Routing (Test account supports 3 topics for 2 stocks: 300750 & 600519)
  natsSubscription *sub_snap_300750 = nullptr;
  natsSubscription *sub_trans_300750 = nullptr;
  natsSubscription *sub_order_300750 = nullptr;
  natsSubscription *sub_snap_600519 = nullptr;
  natsSubscription *sub_trans_600519 = nullptr;
  natsSubscription *sub_order_600519 = nullptr;

  natsConnection_Subscribe(&sub_snap_300750, conn, "stk.l2.300750", onSnapshotMsg, nullptr);
  natsConnection_Subscribe(&sub_trans_300750, conn, "stk.trans.300750", onTransactionMsg, nullptr);
  natsConnection_Subscribe(&sub_order_300750, conn, "stk.order.300750", onOrderMsg, nullptr);

  natsConnection_Subscribe(&sub_snap_600519, conn, "stk.l2.600519", onSnapshotMsg, nullptr);
  natsConnection_Subscribe(&sub_trans_600519, conn, "stk.trans.600519", onTransactionMsg, nullptr);
  natsConnection_Subscribe(&sub_order_600519, conn, "stk.order.600519", onOrderMsg, nullptr);

  // Optimize subscription pending queue bounds to handle high tick-rate
  natsSubscription_SetPendingLimits(sub_snap_300750, 1000000, 64 * 1024 * 1024);
  natsSubscription_SetPendingLimits(sub_trans_300750, 5000000, 256 * 1024 * 1024);
  natsSubscription_SetPendingLimits(sub_order_300750, 5000000, 256 * 1024 * 1024);
  natsSubscription_SetPendingLimits(sub_snap_600519, 1000000, 64 * 1024 * 1024);
  natsSubscription_SetPendingLimits(sub_trans_600519, 5000000, 256 * 1024 * 1024);
  natsSubscription_SetPendingLimits(sub_order_600519, 5000000, 256 * 1024 * 1024);

  SPDLOG_INFO("Subscribed to stock Level 2 topics for 300750 & 600519 (stk.l2, stk.trans, stk.order)");
  SPDLOG_INFO("Listening for messages. Press Ctrl+C to shutdown...");

  // 5. Main event loop
  while (g_running) {
    std::this_thread::sleep_for(std::chrono::milliseconds(200));
  }

  // 6. Graceful cleanup & shutdown
  SPDLOG_INFO("Shutting down client...");

  natsSubscription_Destroy(sub_snap_300750);
  natsSubscription_Destroy(sub_trans_300750);
  natsSubscription_Destroy(sub_order_300750);
  natsSubscription_Destroy(sub_snap_600519);
  natsSubscription_Destroy(sub_trans_600519);
  natsSubscription_Destroy(sub_order_600519);

  natsConnection_Close(conn);
  natsConnection_Destroy(conn);
  nats_Close();

  SPDLOG_INFO("Client closed successfully. Goodbye!");
  spdlog::shutdown();

  return 0;
}