// IM 扇出**背压**回归测试（node:test）—— 2026-09-25 第 15 棒 · 修 `send()` 丢弃 `write()` 返回值。
//
// 修的问题：`send()` 恒 `return true`、不看 `socket.write()` 返回值、也不看水位
// ⇒ **慢客户端（弱网 / 故意不读）让待写缓冲无界增长** ⇒ 单点即可把进程内存顶爆。
//
// 覆盖面：① hub 背压计数（计数 / 峰值 / 断开数）② 扇出语义**未变**（delivered === onlineIn，
//        ⛔ 不因客户端慢而虚减）③ `write()=false` ⇒ 记账但**不丢帧** ④ 水位超限 ⇒ **具名断开**。
//
// ⚠️ 用**假 socket**（`writableLength` 可注入、`write` 可返回 false）⇒ 判据可复现，
//    ⛔ 不必真写 4 MiB 等 OS 缓冲满（那样测试既慢又不稳）。

import { test } from 'node:test'
import assert from 'node:assert/strict'
import { Duplex } from 'node:stream'
import { ImHub } from '../lib/im/hub.js'
import { createImWsRuntime, IM_SEND_BUFFER_LIMIT_BYTES } from '../lib/im/ws.js'

/** 只满足 hub / ws 需要的 store 桩（类型在编译期，运行期只需这几个方法）。 */
function makeStore({ member = true } = {}) {
  return {
    isMember: async () => member,
    findRoom: async () => ({ roomType: 'demo', config: {} }),
    findMember: async () => ({ roomId: 'r1', memberId: 'u1', kind: 'human', role: 'member' }),
    decideWrite: async () => 'ok',
    appendMessage: async () => ({ message: { id: 'm1', seq: 1 }, inserted: true }),
    all: async () => [],
    latestSeq: async () => 0,
    listMessages: async () => [],
    countMembers: async () => 1,
    presenceSnapshot: () => ({}),
  }
}

function makeHub(store) {
  return new ImHub(store, {})
}

/**
 * 假 socket —— 继承 `Duplex`，但把两个关键量**变成可控的**：
 * - `writableLength`：可注入（`setPending`）⇒ 不必真写满 OS 缓冲就能进"超限"分支
 * - `write()` 返回值：`_write` 不回调 `cb` 且 `highWaterMark` 很小 ⇒ 首次写即返回 `false`
 */
class FakeSocket extends Duplex {
  constructor({ highWaterMark = 4 << 20 } = {}) {
    super({ highWaterMark })
    this.written = []
    this.destroyedBySelf = false
    this._pendingOverride = null
    this._blockWrites = false
  }
  /**
   * 注入待写缓冲读数（超限分支用）。
   *
   * ⚠️ 为什么要 `defineProperty` **在实例上**：Node 在构造时把 `writableLength` 定义成
   * **实例自有属性** ⇒ 子类原型上的 getter **被遮蔽、静默失效**（本夹具第一版就踩了这个：
   * 断言"该帧不投递"却投递了，因为读到的仍是真实值 0）。所以必须在**同一层**覆盖。
   */
  setPending(n) {
    this._pendingOverride = n
    Object.defineProperty(this, 'writableLength', { get: () => n, configurable: true })
  }
  /** 让后续 `write()` 返回 false（= 水位满），但不丢数据。 */
  blockWrites() {
    this._blockWrites = true
    this.emit('drain')
  }
  get writableLength() {
    if (this._pendingOverride !== null) return this._pendingOverride
    return super.writableLength
  }
  _write(chunk, _enc, cb) {
    this.written.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    // 阻塞态：**不调 cb** ⇒ Node 认为缓冲未消化 ⇒ 后续 write() 返回 false（数据仍已排队）
    if (!this._blockWrites) cb()
  }
  _read() {}
  destroy(err) {
    this.destroyedBySelf = true
    return super.destroy(err)
  }
  /** 服务端写回的全部字节（拼起来）。 */
  output() {
    return Buffer.concat(this.written).toString('utf8')
  }
}

/** 构造一个**带掩码**的客户端文本帧（RFC 6455 最小必要子集）。 */
function clientTextFrame(text) {
  const payload = Buffer.from(text, 'utf8')
  assert.ok(payload.length < 126, '本夹具只支持 <126 字节的载荷')
  const mask = Buffer.from([1, 2, 3, 4])
  const header = Buffer.from([0x81, 0x80 | payload.length])
  const masked = Buffer.from(payload)
  for (let i = 0; i < masked.length; i += 1) masked[i] ^= mask[i % 4]
  return Buffer.concat([header, mask, masked])
}

/** 伪造一个 upgrade 请求（`ws.ts` 只读 url / headers）。 */
function fakeReq() {
  return {
    url: '/api/im/ws',
    method: 'GET',
    headers: {
      upgrade: 'websocket',
      connection: 'Upgrade',
      'sec-websocket-key': 'dGhlIHNhbXBsZSBub25jZQ==',
      'sec-websocket-version': '13',
      cookie: 'dsh_session=fake',
    },
  }
}

/** 起一套 ws 运行时 + 一个已握手的假 socket（并让它订上 `r1`）。 */
function connected({ limit = 1024, member = true, socketOptions } = {}) {
  const store = makeStore({ member })
  const hub = makeHub(store)
  const logs = []
  const runtime = createImWsRuntime({
    hub,
    store,
    authenticate: async () => ({ userId: 'u1' }),
    sendBufferLimitBytes: limit,
    log: {
      info: (f, m) => logs.push({ level: 'info', f, m }),
      warn: (f, m) => logs.push({ level: 'warn', f, m }),
    },
  })
  const socket = new FakeSocket(socketOptions)
  const handled = runtime.dispatch(fakeReq(), socket, Buffer.alloc(0))
  return { store, hub, runtime, socket, logs, handled }
}

/* ── ① hub 背压计数 ─────────────────────────────────────────────────────── */

test('背压 hub 计数：noteBackpressure 累计 ＋ 取峰值；noteBackpressureClose 单计；stats 带出新字段', () => {
  const hub = makeHub(makeStore())
  assert.equal(hub.stats().backpressureEvents, 0)
  assert.equal(hub.stats().peakPendingBytes, 0)
  assert.equal(hub.stats().closedByBackpressure, 0)

  hub.noteBackpressure(100)
  hub.noteBackpressure(5000)
  hub.noteBackpressure(200) // 峰值不回落
  assert.equal(hub.stats().backpressureEvents, 3)
  assert.equal(hub.stats().peakPendingBytes, 5000)
  // 非法读数**不入峰值**（⛔ 不让 NaN 污染观测面）
  hub.noteBackpressure(Number.NaN)
  assert.equal(hub.stats().peakPendingBytes, 5000)
  assert.equal(hub.stats().backpressureEvents, 4)

  hub.noteBackpressureClose()
  assert.equal(hub.stats().closedByBackpressure, 1)
  // 既有字段一个都没动（⛔ 不是替换而是新增）
  assert.equal(hub.stats().closedByHeartbeat, 0)
  assert.equal(hub.stats().rejectedByQuota, 0)
  assert.equal(hub.stats().deliveries, 0)
})

test('背压 上限常量=4×单帧上限（取值有依据，⛔ 非拍脑袋）', () => {
  assert.equal(IM_SEND_BUFFER_LIMIT_BYTES, 4 << 20)
})

/* ── ② 扇出语义未变（关键回归） ────────────────────────────────────────── */

test('背压 扇出语义**未变**：正常连接 delivered === onlineIn，慢连接也**不虚减** delivered', () => {
  const hub = makeHub(makeStore())
  const sent = []
  const fast = {
    id: 'c1',
    userId: 'u1',
    rooms: new Set(['r1']),
    send: (t) => {
      sent.push(t)
      return true
    },
    close: () => {},
    lastSeenAt: Date.now(),
    openedAt: Date.now(),
  }
  const slow = {
    id: 'c2',
    userId: 'u2',
    rooms: new Set(['r1']),
    // 🔴 慢连接：write() 返回 false（水位满）—— 但**数据已排队、会送达** ⇒ 修复后仍算 delivered
    send: (t) => {
      sent.push(t)
      return true
    },
    close: () => {},
    lastSeenAt: Date.now(),
    openedAt: Date.now(),
  }
  hub.attach(fast)
  hub.attach(slow)
  hub.subscribe(fast, 'r1', false)
  hub.subscribe(slow, 'r1', false)
  assert.equal(hub.onlineIn('r1'), 2)

  const res = hub.deliverFrame('r1', '{"op":"message"}')
  assert.equal(res.delivered, 2)
  assert.equal(res.skipped, 0)
  assert.equal(res.delivered, hub.onlineIn('r1'), '判据 3：delivered 必须等于在线订阅数')
  assert.equal(sent.length, 2)
})

test('背压 真失败（连接已死）仍计入 skipped：⛔ 不与"慢"混为一谈', () => {
  const hub = makeHub(makeStore())
  const dead = {
    id: 'c1',
    userId: 'u1',
    rooms: new Set(['r1']),
    send: () => false, // 连接已死
    close: () => {},
    lastSeenAt: Date.now(),
    openedAt: Date.now(),
  }
  hub.attach(dead)
  hub.subscribe(dead, 'r1', false)
  const res = hub.deliverFrame('r1', 'x')
  assert.equal(res.delivered, 0)
  assert.equal(res.skipped, 1)
})

/* ── ③④ ws 层：write()=false 记账不断开；水位超限 ⇒ 具名断开 ──────────── */

test('背压 ws 握手 + 订阅：dispatch 接管，101 已回，连接进了 hub', async () => {
  const { handled, socket, hub, logs } = connected()
  assert.equal(handled, true, 'dispatch 必须接管 /api/im/ws')
  // ⚠️ 握手是异步的（`authenticate` 是 async）⇒ 必须等一拍再断言
  await new Promise((r) => setTimeout(r, 10))
  assert.match(socket.output(), /HTTP\/1\.1 101/)
  assert.equal(hub.connections.size, 1)
  assert.equal(logs.filter((l) => l.level === 'warn').length, 0)
})

test('背压 write() 返回 false ⇒ 记一次背压，但**仍算 delivered**（数据已排队，⛔ 不丢）', () => {
  // highWaterMark = 1 ⇒ 首次 write 即返回 false（模拟水位满）
  const { socket, hub } = connected({ limit: 8 << 20, socketOptions: { highWaterMark: 1 } })
  // 订上 r1（发一帧 subscribe）
  socket.push(clientTextFrame(JSON.stringify({ op: 'subscribe', roomId: 'r1' })))
  return new Promise((resolve) => setTimeout(resolve, 10)).then(() => {
    socket.blockWrites()
    const res = hub.deliverFrame('r1', 'hello')
    assert.equal(res.delivered, 1, 'write()=false 不等于丢帧 ⇒ 仍算 delivered')
    assert.equal(res.skipped, 0)
    assert.ok(hub.stats().backpressureEvents >= 1, '必须记下这次背压（供观测 / 调阈值）')
    assert.ok(hub.stats().peakPendingBytes > 0)
    // ⛔ 未断开
    assert.equal(hub.stats().closedByBackpressure, 0)
    assert.equal(socket.destroyedBySelf, false)
  })
})

test('背压 🔴 水位超限 ⇒ **具名断开**（1013 send-buffer-overflow）＋ 计数 ＋ 落具名日志', async () => {
  const { socket, hub, logs } = connected({ limit: 1024 })
  socket.push(clientTextFrame(JSON.stringify({ op: 'subscribe', roomId: 'r1' })))
  await new Promise((r) => setTimeout(r, 10))
  assert.equal(hub.onlineIn('r1'), 1)

  // 注入超限水位 ⇒ 下一次 send 必须**拒发并断开**
  socket.setPending(1024 + 1)
  const res = hub.deliverFrame('r1', 'x'.repeat(100))
  assert.equal(res.delivered, 0, '超限 ⇒ 该帧不投递')
  assert.equal(res.skipped, 1)
  assert.equal(hub.stats().closedByBackpressure, 1)
  assert.ok(hub.stats().backpressureEvents >= 1)
  assert.equal(socket.destroyedBySelf, true, '连接必须真被销毁（防内存无界增长）')
  // 断开帧具名（1013 = Try Again Later），且**日志点名原因**
  assert.match(socket.output(), /1013|send-buffer-overflow/)
  const warn = logs.find((l) => l.level === 'warn')
  assert.ok(warn, '必须落一条 warn 日志')
  assert.equal(warn.m, 'im_send_buffer_overflow')
  assert.equal(warn.f.limit, 1024)
  assert.equal(warn.f.pendingBytes, 1025)
})

test('背压 恰好等于上限 ⇒ **不断开**（判据是">limit"，⛔ 不是"≥"）', async () => {
  const { socket, hub } = connected({ limit: 1024 })
  socket.push(clientTextFrame(JSON.stringify({ op: 'subscribe', roomId: 'r1' })))
  await new Promise((r) => setTimeout(r, 10))
  socket.setPending(1024)
  const res = hub.deliverFrame('r1', 'y')
  assert.equal(res.delivered, 1)
  assert.equal(hub.stats().closedByBackpressure, 0)
  assert.equal(socket.destroyedBySelf, false)
})