// Liquidation alert bot: big liquidations and cascades as they happen, plus a warning when the price runs into a large
// liquidation cluster of the heatmap. Three ways to receive events (pick what your plan has):
//   BOT_MODE=stream   live stream (API Pro), reconnects with Last-Event-ID
//   BOT_MODE=webhook  signed webhooks pushed to this process (API Dev, API Pro), verifies Blockmetric-Signature
//   BOT_MODE=poll     polls new liquidations (every key, also the free one)
//   BOT_MODE=auto     stream, falling back to poll without API Pro (default)
//
//   BLOCKMETRIC_API_KEY=bm_live_… BLOCKMETRIC_BASE_URL=https://<domain>/api/v1 node bot.mjs
import http from 'node:http'
import { pathToFileURL } from 'node:url'
import { Blockmetric, BlockmetricError, OPERATIONS, heatmapClusters } from 'blockmetric'
import { constructEvent } from 'blockmetric/webhooks'

export function config(env = process.env) {
  const num = (name, def) => {
    const v = env[name]
    if (v === undefined || v === '') return def
    const n = Number(v)
    if (!Number.isFinite(n)) throw new Error(`${name} must be a number, got "${v}"`)
    return n
  }
  return {
    mode: env.BOT_MODE ?? 'auto',
    coins: (env.BOT_COINS ?? 'BTC,ETH').split(',').map((c) => c.trim()).filter(Boolean),
    kinds: env.BOT_KINDS ?? 'big_liquidation,liq_cascade,whale_liquidated',
    minSeverity: num('BOT_MIN_SEVERITY', 2),
    minUsd: num('BOT_MIN_USD', 250_000),
    clusterMinUsd: num('BOT_CLUSTER_MIN_USD', 5_000_000),
    clusterDistance: num('BOT_CLUSTER_DISTANCE', 0.01),
    clusterMinutes: num('BOT_CLUSTER_MINUTES', 5),
    pollSeconds: num('POLL_SECONDS', 15),
    port: num('PORT', 3000),
    hookPath: env.HOOK_PATH ?? '/hook',
    secret: env.BLOCKMETRIC_WEBHOOK_SECRET,
    alertUrl: env.ALERT_WEBHOOK_URL || null,
    exitAfter: num('BOT_EXIT_AFTER', 0),
  }
}

const usd = (n) => (typeof n === 'number' && Number.isFinite(n) ? `$${n >= 1e9 ? `${(n / 1e9).toFixed(2)}B` : n >= 1e6 ? `${(n / 1e6).toFixed(2)}M` : n >= 1e3 ? `${(n / 1e3).toFixed(1)}k` : n.toFixed(0)}` : '-')
const price = (p) => (typeof p === 'number' ? (p >= 1000 ? p.toLocaleString('en-US', { maximumFractionDigits: 0 }) : p >= 1 ? p.toFixed(2) : p.toPrecision(4)) : '-')

/** One line for a radar event (stream / webhook envelope). */
export const formatEvent = (ev) => `[${ev.data.kind}] sev ${ev.data.severity}${ev.data.coin ? ` ${ev.data.coin}` : ''}: ${ev.data.title}`

/** One line for a single liquidation row of GET /liquidations. */
export const formatLiquidation = (l) => `[liquidation] ${l.coin} ${l.side ?? ''} ${usd(l.usd)} on ${l.venue} at ${price(l.price)} (${l.ts})`

/** Clusters of the newest heatmap column that are large and close to the mark. */
export function nearClusters(heatmap, { minUsd, maxDistance }) {
  const c = heatmapClusters(heatmap, { top: 10 })
  if (!c?.mark) return []
  return [...c.above, ...c.below].filter((x) => x.usd >= minUsd && x.distance !== null && Math.abs(x.distance) <= maxDistance)
}

export async function run(cfg = config(), { client = new Blockmetric({ userAgent: 'liquidation-alert-bot/1.0' }), signal, log = console.log } = {}) {
  const ctrl = new AbortController()
  signal?.addEventListener('abort', () => ctrl.abort(), { once: true })
  let sent = 0
  const seen = new Set() // dedupe: webhooks are at-least-once, clusters stay put for a while

  async function alert(key, text) {
    if (seen.has(key)) return
    seen.add(key)
    log(text)
    if (cfg.alertUrl) {
      // Slack-compatible body; any endpoint that takes JSON works.
      await fetch(cfg.alertUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text, content: text }), signal: AbortSignal.timeout(10_000) }).catch((e) => log(`alert delivery failed: ${e.message}`))
    }
    if (cfg.exitAfter && ++sent >= cfg.exitAfter) ctrl.abort()
  }

  const coins = new Set(cfg.coins.map((c) => c.toUpperCase()))
  const onEvent = (ev) => (ev.data.coin && coins.size && !coins.has(ev.data.coin.toUpperCase()) ? undefined : alert(ev.id, formatEvent(ev)))

  // Heatmap watcher: warn once per cluster when the mark is within clusterDistance of a big cluster.
  async function checkClusters() {
    for (const coin of cfg.coins) {
      try {
        const { data } = await client.heatmap(coin, { tf: '15m', hours: 6 }, { signal: ctrl.signal })
        for (const x of nearClusters(data, { minUsd: cfg.clusterMinUsd, maxDistance: cfg.clusterDistance }))
          await alert(`cluster:${coin}:${x.price}`, `[cluster] ${data.coin}: ${usd(x.usd)} of ${x.side === 'short' ? 'shorts' : 'longs'} liquidate at ${price(x.price)} (${(x.distance * 100).toFixed(2)} % from the mark)`)
      } catch (e) {
        if (!ctrl.signal.aborted) log(`heatmap ${coin}: ${e.message}`)
      }
    }
  }
  let timer = null
  if (cfg.clusterMinutes > 0) {
    void checkClusters()
    timer = setInterval(() => void checkClusters(), cfg.clusterMinutes * 60_000)
  }

  async function poll() {
    // Individual liquidations ≥ minUsd, filtered on the server where the API knows minUsd.
    const serverMin = OPERATIONS.liquidations.query.includes('minUsd')
    log(`polling new liquidations ≥ ${usd(cfg.minUsd)} every ${cfg.pollSeconds}s`)
    const params = { limit: 200, ...(serverMin ? { minUsd: cfg.minUsd } : {}) }
    for await (const l of client.follow('liquidations', params, { intervalMs: cfg.pollSeconds * 1000, signal: ctrl.signal, onError: (e) => log(`poll failed: ${e.message}`) })) {
      if ((l.usd ?? 0) < cfg.minUsd || (coins.size && !coins.has(l.coin.toUpperCase()))) continue
      await alert(`liq:${l.id}`, formatLiquidation(l))
    }
  }

  async function stream() {
    for await (const ev of client.stream(
      { kinds: cfg.kinds, minSeverity: cfg.minSeverity, coin: cfg.coins },
      { signal: ctrl.signal, onReady: () => log('connected to the live stream'), onReconnect: (r) => log(`reconnecting in ${Math.round(r.delayMs / 1000)}s (${r.reason})`) },
    ))
      await onEvent(ev)
  }

  function webhook() {
    if (!cfg.secret) throw new Error('BOT_MODE=webhook needs BLOCKMETRIC_WEBHOOK_SECRET (whsec_…): the signing secret of your endpoint on /account/api.')
    const kinds = new Set(cfg.kinds.split(','))
    const server = http.createServer((req, res) => {
      if (req.method !== 'POST' || req.url?.split('?')[0] !== cfg.hookPath) return void res.writeHead(404).end()
      const chunks = []
      req.on('data', (c) => chunks.push(c))
      req.on('end', async () => {
        let ev
        try {
          ev = await constructEvent(Buffer.concat(chunks), req.headers['blockmetric-signature'], cfg.secret)
        } catch {
          return void res.writeHead(400).end('invalid signature')
        }
        res.writeHead(204).end() // answer first (within 10 s), then work
        if (ev.type === 'test' || (kinds.has(ev.type) && ev.data.severity >= cfg.minSeverity)) await onEvent(ev)
      })
    })
    return new Promise((resolve, reject) => {
      server.on('error', reject)
      server.listen(cfg.port, () => log(`webhook receiver listening on http://localhost:${server.address().port}${cfg.hookPath}`))
      ctrl.signal.addEventListener('abort', () => server.close(() => resolve()), { once: true })
    })
  }

  try {
    if (cfg.mode === 'webhook') await webhook()
    else if (cfg.mode === 'poll') await poll()
    else if (cfg.mode === 'stream') await stream()
    else {
      try {
        await stream()
      } catch (e) {
        if (!(e instanceof BlockmetricError) || (e.code !== 'plan_required' && e.code !== 'scope_required')) throw e
        log('the live stream needs API Pro, falling back to polling')
        await poll()
      }
    }
  } finally {
    if (timer) clearInterval(timer)
  }
}

if (import.meta.url === pathToFileURL(process.argv[1] ?? '').href) {
  const ctrl = new AbortController()
  process.once('SIGINT', () => ctrl.abort())
  process.once('SIGTERM', () => ctrl.abort())
  run(config(), { signal: ctrl.signal }).then(
    () => process.exit(0), // do not wait for idle keep-alive connections
    (e) => {
      console.error(e instanceof BlockmetricError ? `${e.code}: ${e.message}${e.docUrl ? ` (${e.docUrl})` : ''}` : e.message)
      process.exit(1)
    },
  )
}