#include "OrderServiceServer.h"
#include "lib/api/public/ServiceContext.h"
#include "lib/api/public/StatefulPublisher.h"
#include "lib/api/services/AdminService.client.pb.h"
#include "lib/api/services/MarginService.client.pb.h"
#include "lib/api/services/OrderService.pb.h"
#include "lib/api/services/PositionService.client.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/jetfix.pb.h"
#include "lib/config/protos/orderman_cfg.pb.h"
#include "lib/core/FMT.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/common.h"
#include "lib/logging/logging.h"
#include "lib/md/MDClient.h"
#include "lib/md/MDConsumer.h"
#include "lib/md/UserSymbol.h"
#include "lib/mercury/actors/Timer.h"
#include "lib/mercury/executor/Executor.h"
#include "lib/metrics/metrics.h"
#include "lib/orderman/actors/AsyncFactory.h"
#include "lib/orderman/actors/FillStore.h"
#include "lib/orderman/actors/MDSource.h"
#include "lib/orderman/actors/MDSupplementarySource.h"
#include "lib/orderman/actors/OrderStore.h"
#include "lib/orderman/actors/SecuritySrc.h"
#include "lib/orderman/actors/ServiceBridge.h"
#include "lib/orderman/actors/order_manager/OrderManager.h"
#include "lib/orderman/actors/order_manager/OrderServicesData.h"
#include "lib/orderman/actors/order_manager/order_gateway_definition.h"
#include "lib/orderman/events/ExchangePositionEvent.h"
#include "lib/orderman/events/OrderOpResultEvent.h"
#include "lib/orderman/gateways/i_execution_gateway_provider.h"
#include "lib/orderman/order_factory/DynamicOrderRegistry.h"
#include "lib/orderman/orderman_utils/ExchangeOpenOrders.h"
#include "lib/orderman/orders/Order.h"
#include "lib/os/os.h"
#include "lib/os/spark_data_directory.h"
#include "lib/plugin/SparkServices.h"
#include "lib/plugin/plugin.h"
#include "lib/utils/PBUtils.h"
#include "lib/utils/bts_chrono.h"
#include "lib/utils/thread_manager.h"
#include "lib/utils/type_name.h"

#include <bits/stdint-uintn.h>
#include <boost/algorithm/string/predicate.hpp>
#include <functional>
#include <optional>
#include <unordered_map>

namespace bts {
namespace orderman {
using std::unique_ptr;
using std::vector;

struct ActorThread {
   ActorThread(
       const config::OrdermanCfg::Executor &cfg,
       metrics::MetricsContainer &stats)
       : cfg(cfg), thread(), executor(), loop_metric(), done(false) {
      std::array<double, 30> buckets;
      auto i = 0;
      for (auto &b : buckets)
         b = (i += 1000);

      loop_metric = std::make_shared<metrics::HistogramMetric<30>>(
          fmt::format("spark_metrics_{}_thread_loop_ns", cfg.thread_name()),
          buckets);
      stats.publish(loop_metric);
   }

   void start(trace::PerfLog &log) {
      if (cfg.poll())
         thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
            const core::TimeSpan log_interval =
                core::TimeSpan::from_seconds(30);
            core::TimeStamp last_log_time{0};
            core::TimeStamp now;
            core::TimeSpan elapsed;
            std::uint32_t last_count = 0;
            for (;;) {
               now = core::TimeStamp::now();
               elapsed = now - last_log_time;
               if (elapsed > log_interval) {
                  LOG(OrderMan, Info,
                      "OrderServiceServer ActorThread work speed is {:.2f} "
                      "cycles/sec",
                      static_cast<float>(last_count) / elapsed.total_seconds());
                  last_log_time = now;
                  last_count = 0;
               }
               executor.do_work(log);
               last_count++;
            }
         });
      else {
         thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
            // for logging purposes
            const std::string t_name = bts::os::get_thread_name();
            const std::string actors_desc = FMT_JOIN(cfg.hosted_actor(), ", ");
            const core::TimeSpan log_interval =
                core::TimeSpan::from_seconds(30);

            const auto inner_iter = 1000;
            bts::core::TimeStamp last_log_time{0};
            core::TimeStamp now;
            core::TimeSpan elapsed;
            std::uint32_t last_count = 0;
            while (!done) {
               now = core::TimeStamp::now();
               elapsed = now - last_log_time;
               auto log_elapsed = now - last_log_time;
               if (log_elapsed > log_interval) {
                  LOG(OrderMan, Info,
                      "OrderServiceServer ActorThread {} work speed is {:.2f} "
                      "cycles/sec ({})",
                      t_name,
                      static_cast<float>(last_count) / elapsed.total_seconds(),
                      actors_desc);
                  last_log_time = now;
                  last_count = 0;
               }
               auto start = Timer::now();
               for (auto i = 0; i < inner_iter; i++)
                  executor.do_work(log);
               last_count += inner_iter;
               auto stop = Timer::now();
               loop_metric->observe(ns_diff(stop, start) / inner_iter);
               std::this_thread::sleep_for(10ms);
            }
         });
      }
   }

   ~ActorThread() {
      done = true;
      thread.join();
   }

   config::OrdermanCfg::Executor cfg;
   std::thread thread;
   mercury::Executor executor;
   std::shared_ptr<metrics::HistogramMetric<30>> loop_metric;
   std::atomic<bool> done;
};

ActorThread *find_thread_for_actor(
    std::string const &actor_name,
    std::vector<std::unique_ptr<ActorThread>> const &all_executors) {
   for (auto const &ex : all_executors)
      for (auto const &hosted : ex->cfg.hosted_actor())
         if (hosted == "*" || boost::algorithm::to_lower_copy(hosted) ==
                                  boost::algorithm::to_lower_copy(actor_name))
            return ex.get();
   return nullptr;
}

class OpenOrderRecordIDStore {
 public:
   std::uint32_t get_rec_id(const std::string &exch_order_id) {
      auto [it, is_new] = rec_ids_.try_emplace(exch_order_id, rec_id_counter_);
      if (is_new) {
         rec_id_counter_++;
      }
      return it->second;
   }

 private:
   std::uint32_t rec_id_counter_ = 1;
   std::unordered_map<std::string, std::uint32_t> rec_ids_;
};

struct OrderServiceServer::PImpl {
   PImpl(
       bts::api::ServiceContext &ctx,
       const Exchange_positionsStatefulPublisher &exchange_pos_publisher,
       const Exchange_open_ordersStatefulPublisher &open_orders_publisher,
       md::MDClient &mdc,
       DynamicOrderRegistry const &d,
       std::vector<std::unique_ptr<IExecutionGatewayProvider>>
           gateway_providers,
       metrics::MetricsContainer &stats,
       trace::PerfLog &log)
       : context(ctx), service_bridge(),
         gateway_providers(std::move(gateway_providers)), service_ready(false),
         known_orders(), known_fills(), publish_orders(), publish_fills(),
         exchange_pos_publisher(exchange_pos_publisher),
         open_orders_publisher(open_orders_publisher),
         admin_service(ctx.locate<api::admin::AdminService>()),
         position_service(ctx.locate<api::position::PositionService>()),
         margin_service(ctx.locate<api::position::MarginService>()),
         md_consumer(mdc.current_consumer()), stats(stats),
         dyn_order_registry(d), log(log), in_flight_order_ops(),
         current_req_op_id(0) {
      LOG(OrderMan, Info, "OrderServiceServer ctr. n_gateway_providers: {}",
          this->gateway_providers.size());
   }

   bts::api::ServiceContext &context;
   ServiceBridge *service_bridge;

   std::vector<std::unique_ptr<IExecutionGatewayProvider>> gateway_providers;
   std::vector<std::unique_ptr<ActorThread>> actor_threads;

   std::atomic<bool> service_ready;
   std::unordered_set<OrderKey> known_orders;
   std::unordered_set<std::string> known_fills;

   OrderServiceServer::OrdersPublisher publish_orders;
   OrderServiceServer::FillsPublisher publish_fills;
   Exchange_positionsStatefulPublisher exchange_pos_publisher;
   Exchange_open_ordersStatefulPublisher open_orders_publisher;
   api::admin::AdminService admin_service;
   api::position::PositionService position_service;
   api::position::MarginService margin_service;
   bts::md::MDConsumer &md_consumer;
   metrics::MetricsContainer &stats;
   DynamicOrderRegistry const &dyn_order_registry;
   trace::PerfLog &log;
   std::unordered_map<uint32_t, api::Response<api::orderman::Response>>
       in_flight_order_ops;
   uint32_t current_req_op_id;
   OpenOrderRecordIDStore open_order_id_store;
};

OrderServiceServer::OrderServiceServer(
    bts::api::ServiceContext &ctx,
    md::MDClient &mdc,
    DynamicOrderRegistry const &d,
    std::vector<std::unique_ptr<IExecutionGatewayProvider>> gateway_providers,
    metrics::MetricsContainer &stats,
    trace::PerfLog &log,
    md::TradeSummarySignalDispatcher &tssd)
    : api::orderman::OrderServiceBase(ctx),
      me(new PImpl(
          ctx,
          make_exchange_positions_stateful_publisher(),
          make_exchange_open_orders_stateful_publisher(),
          mdc,
          d,
          std::move(gateway_providers),
          stats,
          log)) {
   LOG(md, Info, "OrderServiceServer cons t_id: {} t_name: {}",
       os::get_thread_id(), os::get_thread_name());
   me->publish_orders = make_orders_publisher();
   me->publish_fills = make_fills_publisher();

   std::vector<mercury::Executor *> all_executors;
   for (auto const &ex_cfg : config::get<config::OrdermanCfg>().executor()) {
      auto at = std::make_unique<ActorThread>(ex_cfg, stats);
      all_executors.push_back(&at->executor);
      me->actor_threads.push_back(std::move(at));
   }

   for (auto &t : me->actor_threads)
      t->executor.init_neighbours(all_executors);

   // MDSource
   auto thread = find_thread_for_actor("MDSource", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for MDSource. See the "
               "OrdermanCfg key");

   thread->executor.spawn<MDSource>(me->md_consumer, me->stats, log, &tssd);

   // OrderManager
   thread = find_thread_for_actor("OrderManager", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for OrderManager. See "
               "the OrdermanCfg key");
   {
      thread->executor.spawn<OrderManager>(
          d, me->log, mdc.oracle(), std::move(me->gateway_providers));
   }

   // Timer
   thread = find_thread_for_actor("Timer", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for Timer. See the "
               "OrdermanCfg key");
   thread->executor.spawn<mercury::Timer>();

   // AsyncFactory
   thread = find_thread_for_actor("AsyncFactory", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for AsyncFactory. See "
               "the OrdermanCfg key");
   thread->executor.spawn<AsyncFactory>(d, mdc.oracle());

   // OrderStore
   thread = find_thread_for_actor("OrderStore", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for OrderStore. See "
               "the OrdermanCfg key");
   thread->executor.spawn<OrderStore>(mdc.oracle());

   // FillStore
   thread = find_thread_for_actor("FillStore", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for FilllStore. See "
               "the OrdermanCfg key");
   thread->executor.spawn<FillStore>(mdc.oracle());

   // SecuritySrc
   thread = find_thread_for_actor("SecuritySrc", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for SecuritySrc. See "
               "the OrdermanCfg key");
   thread->executor.spawn<SecuritySrc>(me->md_consumer);

   // MDSupplementarySource
   thread = find_thread_for_actor("MDSupplementarySource", me->actor_threads);
   bts_assert_msg(
       thread,
       "Could not find a configured thread for MDSupplementarySource. See "
       "the OrdermanCfg key");
   thread->executor.spawn<MDSupplementarySource>(me->md_consumer);

   // ServiceBridge
   thread = find_thread_for_actor("ServiceBridge", me->actor_threads);
   bts_assert_msg(
       thread, "Could not find a configured thread for ServiceBridge. See "
               "the OrdermanCfg key");
   me->service_bridge = new ServiceBridge(
       [this]() {
          me->service_ready = true;
          me->context.post([this]() { _do_work(); }, 500);
       },
       [this, &ctx](auto const &orders) {
          ctx.post([this, orders] { this->on_orders(orders); });
       },
       [this, &ctx](auto const &fill) {
          ctx.post([this, fill] { this->on_fills(fill); });
       },
       [this, &ctx](auto const &order_op_result) {
          ctx.post([this, order_op_result] {
             this->on_order_op_result(order_op_result);
          });
       },
       [this, &ctx](const ExchangePosition &exch_pos) {
          LOG(OrderMan, Info, "OrderServiceServer exchange position n_pos: {} ",
              exch_pos.total_exchange_positions.size());
          ctx.post([this, exch_pos] { this->_on_exchange_position(exch_pos); });
       },
       [this, &ctx](const ExchangeOpenOrders &exch_open_orders) {
          LOG(OrderMan, Info,
              "OrderServiceServer open orders n_open_orders: {} ",
              exch_open_orders.all_open_orders.size());
          ctx.post([this, exch_open_orders] {
             this->_on_exchange_open_orders(exch_open_orders);
          });
       },
       me->dyn_order_registry, thread->executor.make_proxy(), mdc.oracle());

   thread->executor.activate(me->service_bridge);

   // spin up threads
   for (auto &t : me->actor_threads)
      t->start(log);

   // wait for the service bridge to be ready
   me->service_bridge->block_until_ready();

   // subscribe to admin model updates and forward them to the service bridge
   me->admin_service.subscribe_admin_data() |
       rxcpp::operators::subscribe<api::admin::Admin_dataUpdate>(
           [this](auto const &update) {
              bts_assert(update.all().size() == 1);
              me->service_bridge->update_admin_data(update.all()[0]);
           });

   // Subscribe to position updates and foward them to the service bridge
   me->position_service.subscribe_pnl_account_data() |
       rxcpp::operators::subscribe<api::position::Pnl_account_dataUpdate>(
           [this](auto const &update) {
              me->service_bridge->update_account_pnl(update);
           });

   // Subscribe
   me->position_service.subscribe_position_data() |
       rxcpp::operators::subscribe<api::position::Position_dataUpdate>(
           [this](auto const &update) {
              me->service_bridge->update_position(update);
           });

   me->margin_service.subscribe_excess_liquidity_report() |
       rxcpp::operators::subscribe<
           api::position::Excess_liquidity_reportUpdate>(
           [this](auto const &update) {
              me->service_bridge->update_excess_liquidity_report(update);
           });
   // We wait long enough for the server to be fully ready. This is not great,
   // we should watch for some signal but right now I am not sure if such a
   // signal exist. We cant do it too early either or some other problem
   // happens. We will come to this later and fix it
   // me->context.post([this]() { _do_work(); }, 300000);
}

OrderServiceServer::~OrderServiceServer() {
   bts_assert(false); // this is badly broken at the moment
}

void OrderServiceServer::on_orders(
    std::vector<api::orderman::OrderRecord> const &orders) {
   LOG(OrderMan, Info, "Received order update for publishing");
   std::vector<api::orderman::OrderRecord> created;
   created.reserve(orders.size());
   std::vector<api::orderman::OrderRecord> updated;
   updated.reserve(orders.size());

   for (auto const &rec : orders) {
      LOG(OrderMan, Diag, "OrderServiceServer::on_orders order: {}",
          utils::pb::to_string(rec));
      auto const &k = OrderKey(rec.key());
      if (me->known_orders.count(k) == 0) {
         me->known_orders.insert(k);
         created.push_back(rec);
      } else
         updated.push_back(rec);
   }

   LOG(OrderMan, Info,
       "OrderServiceServer::on_orders publishing orders to the service API. "
       "created {}, updated {}",
       created.size(), updated.size());

   me->publish_orders(&created, &updated, nullptr);
}

void OrderServiceServer::on_fills(
    std::vector<api::orderman::Fill> const &fills) {
   LOG(OrderMan, Info, "Received fill update for publishing");
   std::vector<api::orderman::Fill> created;
   created.reserve(fills.size());
   std::vector<api::orderman::Fill> updated;
   updated.reserve(fills.size());

   for (auto const &f : fills) {
      auto const &k = f.fill_key();
      if (me->known_fills.count(k) == 0) {
         me->known_fills.insert(k);
         created.push_back(f);
      } else
         updated.push_back(f);
   }
   LOG(OrderMan, Info, "OrderServiceServer::on_fills created: {} updated: {}",
       created.size(), updated.size());
   me->publish_fills(&created, &updated, nullptr);
}

void OrderServiceServer::create_order(
    const api::orderman::Create &c,
    const api::Response<api::orderman::Response> &r) {
   LOG(OrderMan, Info,
       "OrderServiceServer::create_order create: {} pr: {} qty: {}  frc_qty: "
       "{}",
       utils::pb::to_string(c), c.price(), c.quantity(),
       c.has_fractional_quantity() ? c.fractional_quantity().fraction() : 0.0);
   if (!me->service_ready) {
      api::orderman::Response response;
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      r(response);
   } else {
      me->in_flight_order_ops[++me->current_req_op_id] = r;
      me->service_bridge->create_order(me->current_req_op_id, c);
   }
}

void OrderServiceServer::cancel_order(
    const api::orderman::Cancel &cancel,
    const api::Response<api::orderman::Response> &r) {
   if (!me->service_ready) {
      api::orderman::Response response;
      response.set_order_key(cancel.order_key());
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      r(response);
   } else {
      me->in_flight_order_ops[++me->current_req_op_id] = r;
      me->service_bridge->cancel_order(me->current_req_op_id, cancel);
   }
}

void OrderServiceServer::update_order(
    const api::orderman::Update &update,
    const api::Response<api::orderman::Response> &r) {
   LOG(OrderMan, Info, "OrderServiceServer::update_order update: {}",
       utils::pb::to_string(update));
   if (!me->service_ready) {
      api::orderman::Response response;
      response.set_order_key(update.order_key());
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      r(response);
   } else {
      me->in_flight_order_ops[++me->current_req_op_id] = r;
      me->service_bridge->update_order(me->current_req_op_id, update);
   }
}

void OrderServiceServer::detach_order(
    const api::orderman::Detach &detach,
    const api::Response<api::orderman::Response> &r) {
   if (!me->service_ready) {
      api::orderman::Response response;
      response.set_order_key(detach.order_key());
      response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      r(response);
   } else {
      me->in_flight_order_ops[++me->current_req_op_id] = r;
      me->service_bridge->detach_order(me->current_req_op_id, detach);
   }
}

void OrderServiceServer::get_registered_dynamic_order_types(
    const api::VoidMessage &,
    const api::Response<api::orderman::RegisteredDynamicOrderTypes> &r) {
   api::orderman::RegisteredDynamicOrderTypes response;

   auto all_orders = me->dyn_order_registry.get_all_dynamic_order_types();
   for (auto const &desc : all_orders)
      *response.add_dynamic_order_desc() = desc;
   r(response);
}

void OrderServiceServer::manually_bust_all_fills(
    const api::orderman::ManualBustParameters &p,
    const api::Response<api::orderman::Response> &r) {
   LOG(OrderMan, Info, "Received manually_bust_all_fills: {}",
       utils::pb::to_string(p));
   me->service_bridge->manually_bust_all_fills(p);
   api::orderman::Response response;
   r(response);
}

void OrderServiceServer::send_rfq(
    const api::orderman::RFQRequest &rfq,
    const api::Response<api::orderman::Response> &r) {
   LOG(OrderMan, Info, "OrderServiceServer::send_rfq rfq_request: {}",
       utils::pb::to_string(rfq));
   me->service_bridge->send_rfq(rfq);
   api::orderman::Response response;
   r(response);
}

void OrderServiceServer::set_precomputed_theos(
    const bts::api::orderman::PrecomputedTheos &precomp_theos,
    const api::Response<::bts::api::ResultMessage> &reply) {
   LOG(Gen, Info, "OrderServiceServer::set_precomputed_theos count: {}",
       precomp_theos.theos().size());
   me->service_bridge->set_precomputed_theos(precomp_theos);
   bts::api::ResultMessage res = {};
   reply(res);
}

void OrderServiceServer::on_order_op_result(
    OrderOpResultEvent const &op_result) {
   auto it = me->in_flight_order_ops.find(op_result.req_id);
   bts_assert(it != me->in_flight_order_ops.end());
   auto &responder = it->second;
   api::orderman::Response r;
   r.set_order_key(op_result.ok.value());
   r.set_error_code(op_result.result.error);
   r.set_msg(op_result.result.error_msg);
   responder(r);
   me->in_flight_order_ops.erase(it);
}

void OrderServiceServer::_on_exchange_position(
    const ExchangePositionEvent &ev) {
   LOG(OrderMan, Info, "OrderServiceServer::_on_exchange_position n_pos: {}",
       ev.position.total_exchange_positions.size());
   for (const auto &p : ev.position.total_exchange_positions) {
      bts::api::orderman::RetrievedExchangePosition ret_pos = {};
      std::string id = p.exchange_sec_desc + "_" +
                       bts::api::Venue_Name(ev.position.exchange) + "_" +
                       p.account_name;
      ret_pos.set_record_id(id);
      ret_pos.set_security_key(p.security == nullptr ? 0 : p.security->id());
      ret_pos.set_security_resolved(p.security != nullptr);
      ret_pos.set_account_name(p.account_name);
      ret_pos.set_account_id(
          p.account_id == std::nullopt ? 0 : p.account_id.value());
      ret_pos.set_account_resolved(p.account_id != std::nullopt);
      ret_pos.set_exchange_desc(p.exchange_sec_desc);
      ret_pos.set_exchange(ev.position.exchange);
      ret_pos.set_avg_filled_price(p.avg_filled_price.value());
      ret_pos.set_position(p.position);
      me->exchange_pos_publisher.add_or_update(ret_pos);
      LOG(OrderMan, Diag, "OrderServiceServer::_on_exchange_position pos: {}",
          utils::pb::to_string(ret_pos));
   }
   LOG(OrderMan, Info,
       "OrderServiceServer::_on_exchange_position publishing {} records",
       ev.position.total_exchange_positions.size());
   me->exchange_pos_publisher.flush();
}

void OrderServiceServer::_on_exchange_open_orders(
    const ExchangeOpenOrdersEvent &ev) {
   using namespace bts::api::orderman;
   LOG(OrderMan, Info,
       "OrderServiceServer::_on_exchange_open_orders n_open_orders: {} "
       "current_sz: {}",
       ev.open_orders.all_open_orders.size(),
       me->open_orders_publisher.keys().size());
   std::map<uint32_t, RetrievedExchangeOpenOrder> new_state = {};
   auto get_signed_qty = [](const auto &o, Quantity qty) {
      double qty_d = std::abs(qty.full_value());
      if (o.side == api::Side::SIDE_SELL) {
         qty_d = -1.0 * qty_d;
      }
      return qty_d;
   };
   for (const auto &o : ev.open_orders.all_open_orders) {
      RetrievedExchangeOpenOrder o_ord = {};
      auto rec_id = me->open_order_id_store.get_rec_id(o.exchange_order_id);
      o_ord.set_record_id(rec_id);
      o_ord.set_exchange_order_id(o.exchange_order_id);
      o_ord.set_exchange_sec_desc(o.exchange_sec_desc);
      o_ord.set_exchange_account(o.exchange_account);
      o_ord.set_security_key(o.security == nullptr ? 0 : o.security->id());
      o_ord.set_security_resolved(o.security != nullptr);
      o_ord.set_account_id(o.account_id.has_value() ? o.account_id.value() : 0);
      o_ord.set_account_name(o.account_id.has_value() ? o.account_name : "");
      o_ord.set_account_resolved(o.account_id.has_value());
      o_ord.set_tif(o.tif);
      o_ord.set_price(o.price.value());
      o_ord.set_qty(get_signed_qty(o, o.qty));
      o_ord.set_filled_price(o.filled_price.value());
      o_ord.set_filled_qty(get_signed_qty(o, o.filled_qty));
      o_ord.set_creation_timestamp_ms(o.creation_timestamp);
      o_ord.set_last_update_timestamp_ms(o.last_update_timestamp);
      o_ord.set_api(o.api);
      o_ord.set_web(o.web);
      new_state[rec_id] = o_ord;
   }
   const auto &prev_state = me->open_orders_publisher.keys();
   std::vector<RetrievedExchangeOpenOrder const *> added = {};
   std::vector<RetrievedExchangeOpenOrder const *> updated = {};
   std::vector<uint32_t> deleted = {};
   bts::api::StatefulPublisher<RetrievedExchangeOpenOrder, uint32_t>::diff(
       new_state, prev_state, added, updated, deleted);

   for (const auto &k : deleted)
      me->open_orders_publisher.erase(k);
   for (RetrievedExchangeOpenOrder const *o : updated)
      me->open_orders_publisher.add_or_update(*o);
   for (RetrievedExchangeOpenOrder const *o : added)
      me->open_orders_publisher.add_or_update(*o);

   LOG(OrderMan, Info,
       "OrderServiceServer::_on_exchange_open_orders publishing {} records",
       ev.open_orders.all_open_orders.size());
   me->open_orders_publisher.flush();
}

void OrderServiceServer::_do_work() {
   me->service_bridge->do_work();
   me->context.post([this]() { _do_work(); }, 500);
}

} // namespace orderman
} // namespace bts