"use strict";var __importDefault=this&&this.__importDefault||function(e){return e&&e.__esModule?e:{default:e}};Object.defineProperty(exports,"__esModule",{value:!0});exports.metadata=void 0;const exchange_1=__importDefault(require("@b/utils/exchange")),Websocket_1=require("@b/handler/Websocket"),db_1=require("@b/db"),console_1=require("@b/utils/console"),sequelize_1=require("sequelize"),redis_1=require("@b/utils/redis"),utils_1=require("../utils");exports.metadata={};const TICKER_CACHE_KEY="exchange:tickers";class TickerHandler{constructor(){this.accumulatedTickers={};this.tickerInterval=null;this.unblockTime=0;this.isRunning=!1;this.redis=redis_1.RedisSingleton.getInstance()}static getInstance(){TickerHandler.instance||(TickerHandler.instance=new TickerHandler);return TickerHandler.instance}isHandlerRunning(){return this.isRunning}setHandlerRunning(e){this.isRunning=e}startTickerInterval(){this.tickerInterval||(this.tickerInterval=setInterval(this.flushTickers.bind(this),1e3))}stopTickerInterval(){if(this.tickerInterval){clearInterval(this.tickerInterval);this.tickerInterval=null}}async flushTickers(){if(Object.keys(this.accumulatedTickers).length>0){await this.sendTickersToClients(this.accumulatedTickers);await this.updateTickerCache(this.accumulatedTickers);this.accumulatedTickers={}}}async sendTickersToClients(e){Websocket_1.messageBroker.broadcastToSubscribedClients("/api/exchange/ticker",{type:"tickers"},{stream:"tickers",data:e})}async updateTickerCache(e){const t={...await this.getTickerCache(),...e},i=await this.getSymbolsInDB(),s=Object.keys(t).filter(e=>i.includes(e)).reduce((e,i)=>{e[i]=t[i];return e},{});await this.redis.set(TICKER_CACHE_KEY,JSON.stringify(s))}async getTickerCache(){const e=await this.redis.get(TICKER_CACHE_KEY);return e?JSON.parse(e):{}}async getSymbolsInDB(){return(await db_1.models.exchangeMarket.findAll({where:{status:!0},attributes:["currency","pair"],raw:!0})).map(e=>`${e.currency}/${e.pair}`)}async fetchTickersWithRetries(e,t){try{const i=await e.fetchTickers(t);await new Promise(e=>setTimeout(e,1e3));return i}catch(e){await this.disableInvalidMarkets(e,t);throw e}}async watchTickersWithRetries(e,t){try{return await e.watchTickers(t)}catch(e){await this.disableInvalidMarkets(e,t,!0);throw e}}async disableInvalidMarkets(e,t,i=!1){const s=this.extractInvalidSymbols(e.message,t);if(s.length>0){await db_1.models.exchangeMarket.update({status:!1},{where:{[sequelize_1.Op.or]:s.map(e=>{const[t,i]=e.split("/");return{currency:t,pair:i}})}});i&&await exchange_1.default.stopExchange()}}extractInvalidSymbols(e,t){return t.filter(t=>e.includes(t))}processTickers(e,t,i){return i.reduce((i,s)=>{t[s]&&(i[s]={last:t[s].last,baseVolume:t[s].baseVolume||("xt"===e?t[s].info.q:void 0),quoteVolume:t[s].quoteVolume,change:t[s].percentage});return i},{})}async sendInitialTickers(){const e=await this.getTickerCache();await this.sendTickersToClients(e)}async start(){try{this.unblockTime=await(0,utils_1.loadBanStatus)();for(;(0,Websocket_1.hasClients)("/api/exchange/ticker");){if(Date.now()<this.unblockTime){const e=this.unblockTime-Date.now();console.log(`Waiting for ${(0,utils_1.formatWaitTime)(e)} until unblock time`);await new Promise(t=>setTimeout(t,Math.min(e,6e4)));this.unblockTime=await(0,utils_1.loadBanStatus)();continue}const e=await exchange_1.default.startExchange();if(!e){await new Promise(e=>setTimeout(e,5e3));continue}const t=await exchange_1.default.getProvider();try{const i=await this.getSymbolsInDB();if(0===i.length){await new Promise(e=>setTimeout(e,5e3));continue}let s;if("binance"===t){s=await this.fetchTickersWithRetries(e,i);this.startTickerInterval()}else if("kucoin"===t){s=await this.fetchTickersWithRetries(e,i);this.stopTickerInterval()}else if(e&&e.has.watchTickers){s=await this.watchTickersWithRetries(e,i);this.startTickerInterval()}else{s=await this.fetchTickersWithRetries(e,i);this.stopTickerInterval()}const a=this.processTickers(t,s,i);if("binance"===t||"kucoin"!==t&&e.has.watchTickers)Object.assign(this.accumulatedTickers,a);else{await this.sendTickersToClients(a);await this.updateTickerCache(a)}await new Promise(e=>setTimeout(e,1e3))}catch(e){console_1.logger.error("EXCHANGE","Error in ticker loop",e);const t=await(0,utils_1.handleExchangeError)(e,exchange_1.default);if("number"==typeof t){this.unblockTime=t;await(0,utils_1.saveBanStatus)(this.unblockTime)}await new Promise(e=>setTimeout(e,5e3))}}}catch(e){console_1.logger.error("EXCHANGE","Error in ticker handler",e)}finally{this.setHandlerRunning(!1)}}}exports.default=async()=>{const e=TickerHandler.getInstance();await e.sendInitialTickers();if(!e.isHandlerRunning()){e.setHandlerRunning(!0);await e.start()}};