宜博 / vokoworld

import { createCipheriv, createDecipheriv, createHash, createHmac, randomBytes } from 'node:crypto'
import { lookup } from 'node:dns/promises'
import { BlockList, isIP } from 'node:net'

const DELIVERIES = 'vk_webhook_deliveries'
const BATCH = 20
const TIMEOUT_MS = 10_000
const LEASE_MS = 60_000
/** 首次失败后的重试间隔：1 分钟、5 分钟、30 分钟、2 小时；之后只留在 /events 里。 */
const RETRY_AFTER_MS = Object.freeze([60_000, 300_000, 1_800_000, 7_200_000])

// 回环、内网、链路本地（含云元数据 169.254.169.254）、运营商 NAT、组播与保留段一律不投递。
const NON_PUBLIC = new BlockList()
for (const [net, prefix] of [['0.0.0.0', 8], ['10.0.0.0', 8], ['100.64.0.0', 10], ['127.0.0.0', 8], ['169.254.0.0', 16], ['172.16.0.0', 12], ['192.0.0.0', 24], ['192.168.0.0', 16], ['198.18.0.0', 15], ['224.0.0.0', 4], ['240.0.0.0', 4]]) NON_PUBLIC.addSubnet(net, prefix, 'ipv4')
for (const [net, prefix] of [['::', 127], ['fc00::', 7], ['fe80::', 10], ['ff00::', 8], ['64:ff9b::', 96]]) NON_PUBLIC.addSubnet(net, prefix, 'ipv6')

/**
 * @param {string} address IPv4 或 IPv6 字面量
 * @returns {boolean}
 */
export function isPublicAddress(address) {
  const family = isIP(address)
  if (family === 0) throw new Error(`isPublicAddress: not an IP address: ${address}`)
  return !NON_PUBLIC.check(address, family === 4 ? 'ipv4' : 'ipv6')
}

/**
 * `X-VOKO-Signature` 的值：sha256=HMAC(secret, timestamp + "." + body)。
 * @param {string} secret
 * @param {string} timestamp
 * @param {string} body
 * @returns {string}
 */
export function signWebhook(secret, timestamp, body) {
  return `sha256=${createHmac('sha256', secret).update(`${timestamp}.${body}`).digest('hex')}`
}

/**
 * Webhook：Agent 登记 HTTPS 地址后，事件写入时生成投递记录，由 worker 签名投递并按退避重试。
 * Webhook 只是「提醒你来拉」，漏收的事件都能从 /home 或 /events 补到。
 *
 * @param {{ db: import('knex').Knex, secret_key: string }} deps `secret_key` 用于加密存储各 Agent 的签名 secret
 */
export function createWebhooks({ db, secret_key }) {
  if (typeof db !== 'function') throw new Error('createWebhooks: db is required')
  if (typeof secret_key !== 'string' || secret_key === '') throw new Error('createWebhooks: secret_key is required')
  const key = createHash('sha256').update(secret_key).digest()

  /** @param {string} secret */
  function encrypt(secret) {
    const iv = randomBytes(12)
    const cipher = createCipheriv('aes-256-gcm', key, iv)
    const data = Buffer.concat([cipher.update(secret, 'utf8'), cipher.final()])
    return ['v1', iv.toString('base64url'), cipher.getAuthTag().toString('base64url'), data.toString('base64url')].join('.')
  }

  /** @param {string} stored */
  function decrypt(stored) {
    const [version, iv, tag, data] = stored.split('.')
    if (version !== 'v1') throw new Error(`webhook secret has unknown version: ${version}`)
    const decipher = createDecipheriv('aes-256-gcm', key, Buffer.from(iv, 'base64url'))
    decipher.setAuthTag(Buffer.from(tag, 'base64url'))
    return Buffer.concat([decipher.update(Buffer.from(data, 'base64url')), decipher.final()]).toString('utf8')
  }

  /**
   * 只允许 HTTPS，且域名解析出的全部地址都必须是公网地址（防 SSRF）。
   * ponytail: 校验与实际请求各解析一次 DNS，存在重绑定窗口；要堵死就给 fetch 换成固定到已校验 IP 的 dispatcher。
   * @param {string} webhook_url
   * @returns {Promise<string|null>} 不合规时返回原因
   */
  async function urlProblem(webhook_url) {
    const url = new URL(webhook_url)
    if (url.protocol !== 'https:') return 'webhook_url must use https.'
    if (url.username !== '' || url.password !== '') return 'webhook_url must not contain credentials.'
    const host = url.hostname.replace(/^\[|\]$/g, '')
    // 外部 IO 的最小包裹：解析不了的域名是调用方的输入错误，不是程序 bug。
    let addresses
    try {
      addresses = isIP(host) === 0 ? (await lookup(host, { all: true })).map((item) => item.address) : [host]
    } catch {
      return 'webhook_url host cannot be resolved.'
    }
    return addresses.length > 0 && addresses.every(isPublicAddress) ? null : 'webhook_url must resolve to public addresses only.'
  }

  /**
   * 登记或取消（传 null）Webhook。登记成功返回签名 secret，仅此一次。
   * @param {object} _ctx
   * @param {{ agent_id: string, webhook_url: string|null }} input
   * @returns {Promise<{ ok: true, webhook_secret: string|null } | { ok: false, hint: string }>}
   */
  async function register(_ctx, { agent_id, webhook_url }) {
    if (webhook_url === null) {
      await db('vk_agents').where({ id: agent_id }).update({ webhook_url: null, webhook_secret_enc: null, updated_at: new Date() })
      return { ok: true, webhook_secret: null }
    }
    const problem = await urlProblem(webhook_url)
    if (problem !== null) return { ok: false, hint: problem }
    const webhook_secret = randomBytes(32).toString('hex')
    await db('vk_agents').where({ id: agent_id }).update({ webhook_url, webhook_secret_enc: encrypt(webhook_secret), updated_at: new Date() })
    return { ok: true, webhook_secret }
  }

  /**
   * worker：领取并投递一批到期的 Webhook；返回本轮处理的条数。
   * @param {object} _ctx
   * @returns {Promise<number>}
   */
  async function deliverDue(_ctx) {
    const now = new Date()
    const due = await db.transaction(async (trx) => {
      const rows = await trx(DELIVERIES).where({ status: 'pending' }).where('next_attempt_at', '<=', now).orderBy('next_attempt_at', 'asc').limit(BATCH).forUpdate().skipLocked()
      // 把下次尝试时间推后当作租约，避免另一个 worker 同时投递同一条。
      if (rows.length > 0) await trx(DELIVERIES).whereIn('id', rows.map((row) => row.id)).update({ next_attempt_at: new Date(now.getTime() + LEASE_MS) })
      return rows
    })
    for (const delivery of due) {
      const failure = await attempt(delivery)
      if (failure === null) {
        await db(DELIVERIES).where({ id: delivery.id }).update({ status: 'delivered', attempts: delivery.attempts + 1, next_attempt_at: null, delivered_at: new Date() })
        continue
      }
      const retryAfter = RETRY_AFTER_MS[delivery.attempts]
      await db(DELIVERIES).where({ id: delivery.id }).update(retryAfter === undefined
        ? { status: 'failed', attempts: delivery.attempts + 1, next_attempt_at: null }
        : { attempts: delivery.attempts + 1, next_attempt_at: new Date(Date.now() + retryAfter) })
    }
    return due.length
  }

  /**
   * 投递一次。成功返回 null，失败返回原因。
   * @param {{ event_id: number, agent_id: string }} delivery
   * @returns {Promise<string|null>}
   */
  async function attempt(delivery) {
    const [agent, event] = await Promise.all([
      db('vk_agents').where({ id: delivery.agent_id }).first('webhook_url', 'webhook_secret_enc'),
      db('vk_events').where({ id: delivery.event_id }).first('id', 'type', 'task_id', 'created_at'),
    ])
    if (agent === undefined || event === undefined) throw new Error(`vk_webhook_deliveries refers to a missing agent or event, event: ${delivery.event_id}, agent: ${delivery.agent_id}`)
    if (agent.webhook_url === null) return 'webhook was unregistered'
    const problem = await urlProblem(agent.webhook_url)
    if (problem !== null) return problem
    const body = JSON.stringify({ id: String(event.id), type: event.type, created_at: event.created_at.toISOString(), data: { task_id: event.task_id } })
    const timestamp = String(Math.floor(Date.now() / 1000))
    // 对外部 IO 的最小包裹：对方不可达、超时、非 2xx 都是预期失败，按退避重试。
    try {
      const response = await fetch(agent.webhook_url, {
        method: 'POST',
        redirect: 'manual',
        signal: AbortSignal.timeout(TIMEOUT_MS),
        headers: {
          'content-type': 'application/json',
          'x-voko-event': event.type,
          'x-voko-delivery': String(event.id),
          'x-voko-timestamp': timestamp,
          'x-voko-signature': signWebhook(decrypt(agent.webhook_secret_enc), timestamp, body),
        },
        body,
      })
      return response.status >= 200 && response.status < 300 ? null : `receiver returned ${response.status}`
    } catch (err) {
      return err instanceof Error ? err.message : String(err)
    }
  }

  return { register, deliverDue }
}