sporttery/server/models/OddsHistory.js

import mongoose from 'mongoose';
const { Schema } = mongoose;

const TRACKED_IOR_KEYS = [
  'ior_ot_1',
  'ior_ot_2',
  'ior_ot_3',
  'ior_ot_4',
];

const TRACKED_IOR_SET = new Set(TRACKED_IOR_KEYS);
const TRACKED_PLATFORMS = ['pc', 'ob'];
const TRACKED_PLATFORM_SET = new Set(TRACKED_PLATFORMS);
const DEFAULT_PLATFORM = 'pc';

const ODDS_HISTORY_START_OFFSET = 2 * 60 * 60 * 1000;
const ODDS_HISTORY_END_OFFSET = 3 * 60 * 60 * 1000;
const ODDS_HISTORY_RETENTION = 3 * 24 * 60 * 60 * 1000;
const ODDS_HISTORY_CLEANUP_INTERVAL = 5 * 60 * 1000;

const oddsPointSchema = new Schema({
  time: { type: Number, required: true },
  value: { type: Number, required: true },
  source: { type: Number },
  origin: { type: String },
}, { _id: false });

const oddsHistorySchema = new Schema({
  eventId: { type: Number, required: true, unique: true, index: true },
  originId: { type: Number },
  leagueName: { type: String },
  teamHomeName: { type: String },
  teamAwayName: { type: String },
  startTime: { type: Number, required: true, index: true },
  endTime: { type: Number, required: true, index: true },
  markets: {
    type: Map,
    of: [oddsPointSchema],
    default: {},
  },
}, { timestamps: true });

const OddsHistory = mongoose.model('OddsHistory', oddsHistorySchema);

const isTrackedKey = (key) => TRACKED_IOR_SET.has(key);

const normalizePlatform = (platform) => {
  return TRACKED_PLATFORM_SET.has(platform) ? platform : DEFAULT_PLATFORM;
}

const getMarketKey = (platform, ior) => {
  return `${normalizePlatform(platform)}:${ior}`;
}

const parseMarketKey = (key) => {
  if (isTrackedKey(key)) {
    return { ior: key, platform: DEFAULT_PLATFORM };
  }
  const [platform, ior] = String(key).split(/[:.]/);
  if (TRACKED_PLATFORM_SET.has(platform) && isTrackedKey(ior)) {
    return { ior, platform };
  }
  return null;
}

const getMarketEntries = (markets = {}) => {
  if (markets instanceof Map) {
    return Array.from(markets.entries());
  }
  return Object.entries(markets);
}

const normalizeMarkets = (markets = {}) => {
  const normalized = {};
  getMarketEntries(markets).forEach(([key, points]) => {
    const parsed = parseMarketKey(key);
    if (!parsed || !points?.length) {
      return;
    }
    const marketKey = getMarketKey(parsed.platform, parsed.ior);
    normalized[marketKey] = [
      ...(normalized[marketKey] ?? []),
      ...points,
    ].sort((a, b) => a.time - b.time);
  });
  return normalized;
}

const migrateLegacyMarkets = (markets) => {
  getMarketEntries(markets).forEach(([key, points]) => {
    const parsed = parseMarketKey(key);
    if (!parsed || !points?.length) {
      return;
    }
    const marketKey = getMarketKey(parsed.platform, parsed.ior);
    if (key === marketKey) {
      return;
    }
    if (!markets.has(marketKey)) {
      markets.set(marketKey, points);
    }
    else {
      markets.set(marketKey, [
        ...markets.get(marketKey),
        ...points,
      ].sort((a, b) => a.time - b.time));
    }
    markets.delete(key);
  });
}

const isInsideTrackWindow = (eventTime, recordTime = Date.now()) => {
  if (!eventTime) {
    return false;
  }
  return recordTime >= eventTime - ODDS_HISTORY_START_OFFSET &&
    recordTime <= eventTime + ODDS_HISTORY_END_OFFSET;
}

const formatPoint = (event) => {
  const value = event?.v;
  if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) {
    return null;
  }
  return {
    value,
    source: event.s,
    origin: event.r,
  };
}

const appendChangedPoint = (markets, key, point, time) => {
  const history = markets.get(key) ?? [];
  const last = history[history.length - 1];
  if (last && last.value === point.value && last.source === point.source && last.origin === point.origin) {
    return false;
  }
  history.push({ time, ...point });
  markets.set(key, history);
  return true;
}

const recordGameOdds = async ({ game, events, platform = DEFAULT_PLATFORM }) => {
  const eventId = +game?.eventId;
  const startTime = +game?.timestamp;
  const time = Date.now();
  platform = normalizePlatform(platform);

  if (!eventId || !isInsideTrackWindow(startTime, time) || !events) {
    return { updated: false };
  }

  const trackedEntries = Object.entries(events)
    .filter(([key]) => isTrackedKey(key))
    .map(([key, event]) => [key, formatPoint(event)])
    .filter(([, point]) => point);

  const doc = await OddsHistory.findOne({ eventId });
  const markets = doc?.markets ?? new Map();
  migrateLegacyMarkets(markets);
  let changed = false;
  const currentKeys = new Set(trackedEntries.map(([key]) => getMarketKey(platform, key)));

  trackedEntries.forEach(([key, point]) => {
    if (appendChangedPoint(markets, getMarketKey(platform, key), point, time)) {
      changed = true;
    }
  });

  TRACKED_IOR_KEYS.forEach(key => {
    const marketKey = getMarketKey(platform, key);
    if (currentKeys.has(marketKey) || !markets.has(marketKey)) {
      return;
    }
    if (appendChangedPoint(markets, marketKey, { value: 0 }, time)) {
      changed = true;
    }
  });

  if (!changed) {
    return { updated: false };
  }

  await OddsHistory.findOneAndUpdate(
    { eventId },
    {
      $set: {
        originId: game.originId,
        leagueName: game.leagueName,
        teamHomeName: game.teamHomeName,
        teamAwayName: game.teamAwayName,
        startTime,
        endTime: startTime + ODDS_HISTORY_END_OFFSET,
        markets,
      },
    },
    { upsert: true, new: true },
  );

  return { updated: true };
}

const getGameOddsHistory = async (eventId) => {
  if (!eventId) {
    throw new Error('eventId is required');
  }
  const history = await OddsHistory.findOne({ eventId: +eventId }).lean({ flattenMaps: true });
  if (!history) {
    return null;
  }
  const markets = normalizeMarkets(history.markets);
  return {
    ...history,
    markets,
  };
}

const getOddsHistoryGames = async ({ page = 1, pageSize = 50, status = -1, keyword } = {}) => {
  page = Math.max(1, +page || 1);
  pageSize = Math.min(50, Math.max(1, +pageSize || 50));
  status = +status;

  const now = Date.now();
  const query = { eventId: { $gt: 0 } };
  if (status === 1) {
    query.endTime = { $gte: now };
  }
  else if (status === 2) {
    query.endTime = { $lt: now };
  }
  if (keyword) {
    const pattern = new RegExp(String(keyword).trim(), 'i');
    query.$or = [
      { leagueName: pattern },
      { teamHomeName: pattern },
      { teamAwayName: pattern },
    ];
    if (/^\d+$/.test(String(keyword).trim())) {
      query.$or.push({ eventId: +keyword });
    }
  }

  const [histories, total] = await Promise.all([
    OddsHistory.find(query)
    .select('eventId leagueName teamHomeName teamAwayName startTime endTime markets updatedAt')
    .sort({ startTime: -1 })
    .skip((page - 1) * pageSize)
    .limit(pageSize)
    .lean({ flattenMaps: true }),
    OddsHistory.countDocuments(query),
  ]);

  const list = histories.map(history => ({
    eventId: history.eventId,
    leagueName: history.leagueName,
    teamHomeName: history.teamHomeName,
    teamAwayName: history.teamAwayName,
    startTime: history.startTime,
    endTime: history.endTime,
    marketCount: Object.keys(normalizeMarkets(history.markets)).length,
    updatedAt: history.updatedAt,
  }));

  return { list, page, pageSize, total };
}

const cleanupExpiredHistory = async () => {
  const expireTime = Date.now() - ODDS_HISTORY_RETENTION;
  return OddsHistory.deleteMany({ startTime: { $lt: expireTime } });
}

const startCleanup = (logger = console) => {
  const cleanup = () => {
    cleanupExpiredHistory()
    .then(result => {
      if (result.deletedCount) {
        logger.out?.('odds history cleanup deleted %d records', result.deletedCount);
      }
    })
    .catch(err => {
      logger.out?.('odds history cleanup failed: %s', err.message);
    });
  };

  cleanup();
  return setInterval(cleanup, ODDS_HISTORY_CLEANUP_INTERVAL);
}

export {
  getGameOddsHistory,
  getOddsHistoryGames,
  recordGameOdds,
  startCleanup,
};

export default {
  getGameOddsHistory,
  getOddsHistoryGames,
  recordGameOdds,
  startCleanup,
};