ppai/server/models/Platforms.js

import { fork } from "child_process";
import Store from "../state/store.js";
import ProcessData from "../libs/processData.js";
import Logs from "../libs/logs.js";

import { getGamesRelationsMap } from "../libs/getGamesRelations.js";
import { getSoccerGames, getObossOdds } from "./Partner.js";

const getChildOptions = (inspect=9230) => {
  return process.env.NODE_ENV == 'development' ? {
    execArgv: [`--inspect=${inspect}`],
    stdio: ['pipe', 'pipe', 'pipe', 'ipc'],
    env: { ...process.env }
  } : {
    stdio: ['pipe', 'pipe', 'pipe', 'ipc'],
    env: { ...process.env }
  };
}

const triangleProcess = fork("triangle/main.js", [], getChildOptions(9229));
const triangleData = new ProcessData(triangleProcess, 'triangle');

triangleData.registerResponse('gamesRelations', async () => {
  const gamesRelations = Object.values(getGamesRelationsMap(true));
  return Promise.resolve(gamesRelations);
});

triangleData.registerRequest('solutions', solutions => {
  const oldSolutions = new Map((Store.get('solutions') ?? []).map(item => [item.sid, item]));
  const newSolutions = new Map(solutions.map(item => [item.sid, item]));
  const changed = {
    add: [],
    update: [],
    remove: [],
  }
  oldSolutions.forEach((item, sid) => {
    if (!newSolutions.has(sid)) {
      changed.remove.push(sid);
    }
    else if (newSolutions.get(sid).sol.win_profit_rate != item.sol.win_profit_rate || JSON.stringify(newSolutions.get(sid).cpr) != JSON.stringify(item.cpr)) {
      changed.update.push(sid);
    }
  });
  newSolutions.forEach((item, sid) => {
    if (!oldSolutions.has(sid)) {
      changed.add.push(sid);
    }
  });
  if (changed.update.length || changed.add.length || changed.remove.length) {
    Store.set('solutions', solutions);
    const profitableSolutions = solutions.filter(solution => solution.sol.win_profit_rate > 0);
    Logs.outDev('profitable solutions', profitableSolutions);
  }
});

/**
 * 通用的平台数据更新函数
 * @param {string} platform - 平台名称
 * @param {Array} newItems - 新的数据项数组
 * @param {string} storeKey - Store 中的键名
 * @returns {Promise}
 */
const updatePlatformData = async ({ platform, newItems, storeKey }) => {
  if (!platform || !newItems?.length) {
    return Promise.reject(new Error('invalid request', { cause: 400 }));
  }

  let changed = false;
  const storeData = Store.get(storeKey) ?? {};
  const { [platform]: storePlatformItems = [] } = storeData;
  const storePlatformItemsMap = new Map(storePlatformItems.map(item => [item.id, item]));
  const newPlatformItemsMap = new Map(newItems.map(item => [item.id, item]));

  // 删除不存在的项
  storePlatformItemsMap.forEach(item => {
    if (!newPlatformItemsMap.has(item.id)) {
      storePlatformItemsMap.delete(item.id);
      changed = true;
    }
  });

  // 添加新的项
  newPlatformItemsMap.forEach(item => {
    if (!storePlatformItemsMap.has(item.id)) {
      storePlatformItemsMap.set(item.id, item);
      changed = true;
    }
  });

  // 更新 Store 中的数据
  if (changed) {
    const updatedPlatformItems = Array.from(storePlatformItemsMap.values());
    storeData[platform] = updatedPlatformItems;
    Store.set(storeKey, storeData);
  }

  return Promise.resolve();
};

/**
 * 更新联赛数据
 * @param {string} platform - 平台名称
 * @param {Array} leagues - 联赛数据
 * @returns
 */
export const updateLeagues = async ({ platform, leagues }) => {
  return updatePlatformData({ platform, newItems: leagues, storeKey: 'leagues' });
};

/**
 * 获取过滤后的联赛数据
 * @param {string} platform - 平台名称
 * @returns
 */
export const getRelatedLeagues = async (platform) => {
  const polymarketLeagues = Store.get('polymarket', 'leagues') ?? [];
  const polymarketLeaguesSet = new Set(polymarketLeagues.map(item => item.id));
  const leaguesRelations = Store.get('leaguesRelations') ?? {};
  const filteredLeagues = Object.values(leaguesRelations).filter(relation => {
    return polymarketLeaguesSet.has(relation.platforms.polymarket.id);
  }).map(relation => relation.platforms[platform]);
  return filteredLeagues;
}

/**
 * 更新比赛数据
 * @param {string} platform - 平台名称
 * @param {Array} games - 比赛数据
 * @returns
 */
export const updateGames = async ({ platform, games }) => {
  return updatePlatformData({ platform, newItems: games, storeKey: 'games' });
};

/**
 * 获取过滤后的比赛数据
 * @param {string} platform - 平台名称
 * @returns
 */
export const getRelatedGames = async (platform) => {
  const gamesRelations = Store.get('gamesRelations') ?? {};
  const filteredGames = Object.values(gamesRelations).map(relation => relation.platforms[platform]);
  return filteredGames;
}

/**
 * 更新赔率数据
 * @param {string} platform - 平台名称
 * @param {Array} games - 赔率数据
 * @param {number} timestamp - 时间戳
 * @returns
 */
export const updateOdds = async ({ platform, games, timestamp }) => {
  Store.set(platform, { games, timestamp }, 'odds');
  return Promise.resolve();
};

/**
 * 同步QBoss赔率数据
 * @param {*} relationsData
 */
const syncObossOdds = (relationsData) => {
  const timestamp = Date.now();
  const pinnacle = [];
  const obsports = [];
  const huangguan = [];
  relationsData.forEach(item => {
    Object.entries(item.rel).forEach(([key, value]) => {
      const { eventId: id, ...rest } = value;
      const game = { id, ...rest }
      if (key === 'pc') {
        pinnacle.push(game);
      }
      else if (key === 'ob') {
        obsports.push(game);
      }
      else if (key === 'hg') {
        huangguan.push(game);
      }
    });
  });
  Promise.all([
    updateOdds({ platform: 'pinnacle', games: pinnacle, timestamp }),
    updateOdds({ platform: 'obsports', games: obsports, timestamp }),
    updateOdds({ platform: 'huangguan', games: huangguan, timestamp }),
  ]);
}

/**
 * 定时更新QBoss赔率数据
 */
const updateObossOdds = () => {
  getObossOdds()
  .then(res => {
    if (res.statusCode === 200) {
      // Logs.outDev('update oboss odds', res.data);
      return syncObossOdds(res.data);
    }
    return Promise.reject(new Error(`status code ${res.statusCode}`));
  })
  .catch(error => {
    Logs.err('failed to update oboss odds', error.message);
  })
  .finally(() => {
    setTimeout(() => {
      updateObossOdds();
    }, 1000 * 2);
  });
}
updateObossOdds();

/**
 * 同步关联比赛数据
 * @param {*} relationsData
 */
const syncGamesRelations = (relationsData) => {
  const storeRelations = Store.get('gamesRelations') ?? {};
  const newRelations = relationsData.map(item => {
    const {
      event_id: pmId,
      ps_event_id: pcId,
      ob_event_id: obId,
      hg_event_id: hgId,
      start_time: startTime,
    } = item;
    const timestamp = new Date(startTime).getTime();
    return {
      id: pmId,
      timestamp,
      platforms: {
        polymarket: { id: +pmId },
        pinnacle: { id: +pcId },
        huangguan: { id: +hgId },
        obsports: { id: +obId }
      }
    }
  });
  const changed = { add: 0, update: 0, remove: 0 }
  const newRelationsSet = new Set(newRelations.map(item => item.id));
  Object.keys(storeRelations).forEach(id => {
    if (!newRelationsSet.has(+id)) {
      delete storeRelations[id];
      changed.remove++;
    }
  });
  newRelations.forEach(item => {
    if (!storeRelations[item.id]) {
      storeRelations[item.id] = item;
      changed.add++;
    }
  });

  if (changed.add || changed.update || changed.remove) {
    Store.set('gamesRelations', storeRelations);
    Logs.outDev('sync games relations', changed, storeRelations);
  }
}

/**
 * 定时更新关联比赛
 */
const updateGamesRelations = () => {
  getSoccerGames()
  .then(res => {
    if (res.success) {
      return syncGamesRelations(res.data);
    }
    return Promise.reject(new Error(res.message));
  })
  .catch(error => {
    Logs.err('failed to update games relations', error.message);
  })
  .finally(() => {
    setTimeout(() => {
      updateGamesRelations();
    }, 1000 * 30);
  });
}
updateGamesRelations();

export default {
  updateLeagues, getRelatedLeagues,
  updateGames, getRelatedGames,
  updateOdds,
};