process.env.DB_HOST = '127.0.0.1'
process.env.DB_PORT = '59999'

const { test, after, afterEach, beforeEach } = require('node:test')
const assert = require('node:assert/strict')
const { EventEmitter } = require('node:events')

// The SSE fan-out is the security boundary (docs/link/v3.md §3.6). Before v3 it
// was a static kind allowlist; now each subscriber carries the audience rung it
// resolved to at subscribe time, and every frame is gated + field-projected per
// viewer. These tests pin the properties that must hold no matter how the config
// is set:
//
//   - the admin channel always gets the frame verbatim;
//   - a public subscriber never receives an unmapped kind;
//   - acct / webId never reach a public subscriber, at any rung below admin;
//   - two subscribers at different rungs get different frames from one event;
//   - a viewer's rung is frozen at subscribe time, not re-read per frame;
//   - if the visibility config can't be read, nothing goes out on public.

const broadcast = require('../src/utils/shardBroadcast')
const visibility = require('../src/utils/shardVisibility')
const model = require('../src/model/shardVisibility/shardVisibility.model')
const shardLinks = require('../src/model/shardLinks/shardLinks.model')
const db = require('../src/utils/db')

after(() => db.close())

const originals = {
  listAll: model.listAll,
  listForUser: shardLinks.listForUser,
  getConfig: visibility.getConfig,
  viewerLevel: visibility.viewerLevel,
}

beforeEach(() => {
  model.listAll = async () => []
  shardLinks.listForUser = async () => []
  visibility.invalidate()
})

afterEach(() => {
  broadcast.closeAll()
  model.listAll = originals.listAll
  shardLinks.listForUser = originals.listForUser
  visibility.getConfig = originals.getConfig
  visibility.viewerLevel = originals.viewerLevel
  visibility.invalidate()
})

// A fake req/res pair that records everything written to the stream.
function fakeClient() {
  const req = new EventEmitter()
  const writes = []
  const res = {
    writeHead() {},
    write(chunk) {
      writes.push(chunk)
    },
    end() {},
    on() {},
  }
  // Frames only — drop the SSE comments/retry preamble and keepalive pings.
  const frames = () =>
    writes
      .filter((w) => w.startsWith('data: '))
      .map((w) => JSON.parse(w.slice('data: '.length).trim()))
  return { req, res, frames }
}

async function subscribeAt(level, channel = 'public') {
  const client = fakeClient()
  visibility.viewerLevel = async () => level
  await broadcast.subscribe(client.req, client.res, channel)
  return client
}

const GUILD_FRAME = {
  kind: 'guild.update',
  id: 7,
  name: 'The Nameless',
  abbr: 'TN',
  leader: { serial: '0x1A2B', name: 'Darrow', acct: 'whitlocktech', webId: '42', player: true },
}

test('the admin channel receives the frame verbatim, acct and webId included', async () => {
  const admin = await subscribeAt('admin', 'admin')
  await broadcast.broadcast(GUILD_FRAME)
  const [frame] = admin.frames()
  assert.deepEqual(frame, GUILD_FRAME)
})

test('a public subscriber never sees acct or webId, at any rung below admin', async () => {
  for (const level of ['anonymous', 'logged_in', 'player', 'staff']) {
    const client = await subscribeAt(level)
    await broadcast.broadcast(GUILD_FRAME)
    const [frame] = client.frames()
    assert.ok(frame, `${level} should receive the guild frame`)
    assert.equal(frame.leader.name, 'Darrow')
    assert.equal('acct' in frame.leader, false, `${level} must not see acct`)
    assert.equal('webId' in frame.leader, false, `${level} must not see webId`)
    broadcast.closeAll()
  }
})

test('an unmapped kind reaches the admin channel and nobody else', async () => {
  const anon = await subscribeAt('anonymous')
  const staff = await subscribeAt('staff')
  const admin = await subscribeAt('admin', 'admin')

  for (const kind of ['audit.command', 'cheat.fastwalk', 'account.login.attempt', 'vendor.sale']) {
    await broadcast.broadcast({ kind, secret: true })
  }

  assert.deepEqual(anon.frames(), [])
  assert.deepEqual(staff.frames(), [])
  assert.equal(admin.frames().length, 4)
})

test('one event yields different frames for subscribers at different rungs', async () => {
  model.listAll = async () => [
    {
      feature: 'guilds',
      enabled: true,
      audience: 'anonymous',
      stream: true,
      fieldRules: { abbr: 'staff' },
    },
  ]
  visibility.invalidate()

  const anon = await subscribeAt('anonymous')
  const staff = await subscribeAt('staff')
  await broadcast.broadcast(GUILD_FRAME)

  assert.equal('abbr' in anon.frames()[0], false)
  assert.equal(staff.frames()[0].abbr, 'TN')
  // Both still lose the locked fields.
  assert.equal('acct' in staff.frames()[0].leader, false)
})

test('raising a feature audience cuts off the lower rungs mid-stream', async () => {
  const anon = await subscribeAt('anonymous')
  const player = await subscribeAt('player')

  await broadcast.broadcast(GUILD_FRAME)
  assert.equal(anon.frames().length, 1)
  assert.equal(player.frames().length, 1)

  // Config changes DO take effect live — only the viewer's rung is frozen.
  model.listAll = async () => [
    { feature: 'guilds', enabled: true, audience: 'player', stream: true, fieldRules: {} },
  ]
  visibility.invalidate()

  await broadcast.broadcast(GUILD_FRAME)
  assert.equal(anon.frames().length, 1, 'anonymous stops receiving')
  assert.equal(player.frames().length, 2, 'player keeps receiving')
})

test("a subscriber's rung is frozen at subscribe time", async () => {
  const client = await subscribeAt('anonymous')
  // Even if the resolver would now say "admin", the open connection must not
  // gain privilege — its level was captured when it subscribed.
  visibility.viewerLevel = async () => 'admin'
  await broadcast.broadcast({ kind: 'audit.command', command: 'ban' })
  assert.deepEqual(client.frames(), [])
})

test('an unresolvable viewer subscribes as anonymous, not as privileged', async () => {
  const client = fakeClient()
  visibility.viewerLevel = async () => {
    throw new Error('session lookup exploded')
  }
  await broadcast.subscribe(client.req, client.res, 'public')
  await broadcast.broadcast({ kind: 'audit.command', command: 'ban' })
  assert.deepEqual(client.frames(), [])

  // ...but it still receives ordinary public traffic.
  await broadcast.broadcast({ kind: 'champ.update', serial: '0x1' })
  assert.equal(client.frames().length, 1)
})

test('an unreadable visibility config withholds every public frame', async () => {
  const client = await subscribeAt('anonymous')
  visibility.getConfig = async () => {
    throw new Error('db down')
  }
  await broadcast.broadcast(GUILD_FRAME)
  assert.deepEqual(client.frames(), [])
})

test('a disabled feature stops its kinds without touching others', async () => {
  model.listAll = async () => [
    { feature: 'champs', enabled: false, audience: 'anonymous', stream: true, fieldRules: {} },
  ]
  visibility.invalidate()

  const client = await subscribeAt('anonymous')
  await broadcast.broadcast({ kind: 'champ.update', serial: '0x1' })
  await broadcast.broadcast({ kind: 'guild.update', id: 7 })

  const kinds = client.frames().map((f) => f.kind)
  assert.deepEqual(kinds, ['guild.update'])
})

test('a dead client is dropped rather than repeatedly retried', async () => {
  const client = fakeClient()
  visibility.viewerLevel = async () => 'anonymous'
  await broadcast.subscribe(client.req, client.res, 'public')
  assert.equal(broadcast.stats().publicClients, 1)

  client.res.write = () => {
    throw new Error('EPIPE')
  }
  await broadcast.broadcast({ kind: 'champ.update', serial: '0x1' })
  assert.equal(broadcast.stats().publicClients, 0)
})

test('broadcast is a no-op for a malformed event', async () => {
  const client = await subscribeAt('anonymous')
  await broadcast.broadcast(null)
  await broadcast.broadcast({})
  assert.deepEqual(client.frames(), [])
})