Cascade and cluster alerts with webhooks

// Cascade and cluster alerts: receives signed Blockmetric webhooks (liquidation cascades, big liquidations) and warns when
// the price comes close to a large liquidation cluster of the heatmap. Node 18+, no dependencies.
//
//   export BLOCKMETRIC_WEBHOOK_SECRET=whsec_...       # the endpoint's signing secret from Account > API keys
//   export BLOCKMETRIC_API_KEY=bm_live_...  BLOCKMETRIC_BASE_URL=https://<domain>/api/v1   # for the cluster check
//   export ALERT_WEBHOOK_URL=https://hooks.slack.com/...   # optional: Slack incoming webhook
//   node receiver.mjs                                   # listens on http://localhost:3000/hooks/blockmetric
//
// Options: PORT (3000), COINS (BTC,ETH), CLUSTER_MIN_USD (5000000), CLUSTER_DISTANCE (0.02 = 2 %), CLUSTER_MINUTES (5),
// SELF_TEST=1 (sends itself one signed test event, checks it and exits), ONCE=1 (one cluster check, then exit).
import crypto from 'node:crypto'
import http from 'node:http'

const SECRET = process.env.BLOCKMETRIC_WEBHOOK_SECRET ?? ''
const BASE = (process.env.BLOCKMETRIC_BASE_URL ?? '').replace(/\/+$/, '')
const KEY = process.env.BLOCKMETRIC_API_KEY
const PORT = Number(process.env.PORT ?? 3000)
const COINS = (process.env.COINS ?? 'BTC,ETH').split(',').map((c) => c.trim()).filter(Boolean)
const CLUSTER_MIN_USD = Number(process.env.CLUSTER_MIN_USD ?? 5_000_000)
const CLUSTER_DISTANCE = Number(process.env.CLUSTER_DISTANCE ?? 0.02)
const CLUSTER_MINUTES = Number(process.env.CLUSTER_MINUTES ?? 5)
const usd = (n) => (n >= 1e9 ? `$${(n / 1e9).toFixed(2)}B` : n >= 1e6 ? `$${(n / 1e6).toFixed(1)}M` : `$${(n / 1e3).toFixed(0)}K`)

/** Blockmetric-Signature: t=<unix>,v1=<hex HMAC-SHA256(secret, t + "." + raw body)>[,v1=...]. */
export function verify(rawBody, header, secret, nowSec = Math.floor(Date.now() / 1000)) {
  let t = NaN
  const sigs = []
  for (const part of String(header ?? '').split(',')) {
    const [k, v] = part.trim().split('=')
    if (k === 't') t = Number(v)
    if (k === 'v1' && v) sigs.push(v)
  }
  if (!Number.isInteger(t) || Math.abs(nowSec - t) > 300) return false // replay protection: 5 minutes
  const expected = crypto.createHmac('sha256', secret).update(`${t}.`).update(rawBody).digest()
  return sigs.some((s) => {
    const got = Buffer.from(s, 'hex')
    return got.length === expected.length && crypto.timingSafeEqual(got, expected)
  })
}

const sent = new Set() // event ids and cluster keys already alerted (use your database in production)
async function alert(key, text) {
  if (sent.has(key)) return
  sent.add(key)
  console.log(text)
  const url = process.env.ALERT_WEBHOOK_URL
  if (url) await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text, content: text }) }).catch((e) => console.error(`alert failed: ${e.message}`))
}

const server = http.createServer((req, res) => {
  if (req.method !== 'POST' || req.url !== '/hooks/blockmetric') return void res.writeHead(404).end()
  const chunks = []
  req.on('data', (c) => chunks.push(c))
  req.on('end', () => {
    const raw = Buffer.concat(chunks) // the exact bytes: never parse and re-serialise before verifying
    if (!verify(raw, req.headers['blockmetric-signature'], SECRET)) return void res.writeHead(400).end('invalid signature')
    res.writeHead(204).end() // answer within 10 s, work afterwards
    const ev = JSON.parse(raw.toString('utf8'))
    const e = ev.data
    void alert(ev.id, `[${ev.type}] severity ${e.severity}${e.coin ? ` ${e.coin}` : ''}: ${e.title}`)
  })
})

/** Large clusters of the newest heatmap column that the mark is close to (data.clusters, no grid parsing needed). */
async function checkClusters() {
  for (const coin of COINS) {
    const res = await fetch(`${BASE}/heatmap/${encodeURIComponent(coin)}?tf=15m&hours=1&clusters=10`, { headers: KEY ? { Authorization: `Bearer ${KEY}` } : { 'X-Blockmetric-Access': 'public' } })
    const body = await res.json()
    if (!res.ok) {
      console.error(`heatmap ${coin}: ${body.error?.code} ${body.error?.message}`)
      continue
    }
    const c = body.data.clusters
    if (!c) continue
    const near = [...c.above, ...c.below].filter((x) => x.usd >= CLUSTER_MIN_USD && Math.abs(x.distance) <= CLUSTER_DISTANCE)
    for (const x of near) await alert(`cluster:${coin}:${x.price}`, `[cluster] ${coin}: ${usd(x.usd)} of ${x.shortUsd > x.longUsd ? 'shorts' : 'longs'} liquidate at ${x.price.toLocaleString('en-US')} (${(x.distance * 100).toFixed(2)} % from the mark ${c.mark.toLocaleString('en-US')})`)
    if (!near.length) {
      const nearest = [...c.above, ...c.below].filter((x) => x.usd >= CLUSTER_MIN_USD).sort((a, b) => Math.abs(a.distance) - Math.abs(b.distance))[0]
      console.log(`[cluster] ${coin}: no large cluster within ${CLUSTER_DISTANCE * 100} %${nearest ? `, nearest ${usd(nearest.usd)} at ${nearest.price.toLocaleString('en-US')} (${(nearest.distance * 100).toFixed(1)} %)` : ''}`)
    }
  }
}

if (process.env.ONCE === '1') {
  await checkClusters()
} else {
  if (!SECRET.startsWith('whsec_')) throw new Error('Set BLOCKMETRIC_WEBHOOK_SECRET (whsec_...), the signing secret of your endpoint.')
  server.listen(PORT, () => console.log(`Listening on http://localhost:${PORT}/hooks/blockmetric`))
  if (BASE && CLUSTER_MINUTES > 0) {
    void checkClusters()
    setInterval(() => void checkClusters().catch((e) => console.error(e.message)), CLUSTER_MINUTES * 60_000)
  }
  if (process.env.SELF_TEST === '1') {
    // A signed test delivery to ourselves, exactly as Blockmetric signs it: proves the secret and the raw-body handling.
    const body = JSON.stringify({ id: `evt_test_${crypto.randomBytes(4).toString('hex')}`, type: 'test', created: Math.floor(Date.now() / 1000), data: { severity: 1, coin: null, title: 'Test event: your receiver verifies signatures' } })
    const t = Math.floor(Date.now() / 1000)
    const sig = crypto.createHmac('sha256', SECRET).update(`${t}.${body}`).digest('hex')
    const ok = await fetch(`http://localhost:${PORT}/hooks/blockmetric`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'Blockmetric-Signature': `t=${t},v1=${sig}` }, body })
    const bad = await fetch(`http://localhost:${PORT}/hooks/blockmetric`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'Blockmetric-Signature': `t=${t},v1=${'0'.repeat(64)}` }, body })
    console.log(`self-test: valid signature answered ${ok.status}, forged signature answered ${bad.status}`)
    setTimeout(() => process.exit(ok.status === 204 && bad.status === 400 ? 0 : 1), 200)
  }
}