/*
 * ★ Two tenants, one twin name — each subscriber receives its own tenant's messages and nothing else.
 *
 * The same instanceId (and the same reference source) in two tenants is the expected case: source
 * names such as `sap-ewm-1710` are vocabulary tenants share (`runtime-key.ts`). Until 2026-09-24 both
 * channels filtered on that name alone, so tenant B's subscriber received A's deltas, and the snapshot
 * sent when A subscribed went to B as well.
 *
 * This counts what arrives. A filter test alone would not see a sender that builds its own payload
 * without a domain, so the last two tests read the source: every sender goes through the channel's
 * payload builder, and nothing else names the topic.
 */
import { test } from 'node:test'
import assert from 'node:assert/strict'
import { readdirSync, readFileSync, statSync } from 'node:fs'
import { dirname, join, relative } from 'node:path'
import { fileURLToPath } from 'node:url'

import { TWIN_STATE_TOPIC, twinStateFilter, twinStatePayload } from '../server/engine/twin-state-channel.ts'
import {
  REFERENCE_PROGRESS_TOPIC,
  referenceProgressFilter,
  referenceProgressPayload
} from '../server/service/reference/reference-progress-channel.ts'

const A = 'domain-a'
const B = 'domain-b'
const SHARED = 'sap-ewm-1710'

/* A pubsub reduced to what matters here: every published payload, and a subscriber's filter over them. */
function bus() {
  const sent: { topic: string; payload: any }[] = []
  return {
    publish: (topic: string, payload: any) => void sent.push({ topic, payload }),
    received: (topic: string, keep: (p: any) => boolean) => sent.filter(m => m.topic === topic).map(m => m.payload).filter(keep)
  }
}

test('twin-state: B receives none of A\'s snapshot or deltas for a twin of the same name', () => {
  const b = bus()
  /* A subscribes (its snapshot goes out), then both tenants' twins tick. */
  b.publish(TWIN_STATE_TOPIC, twinStatePayload(A, { instanceId: SHARED, kind: 'snapshot', revision: 7, payload: { owner: 'A' } }))
  b.publish(TWIN_STATE_TOPIC, twinStatePayload(A, { instanceId: SHARED, kind: 'delta', revision: 8 }))
  b.publish(TWIN_STATE_TOPIC, twinStatePayload(B, { instanceId: SHARED, kind: 'delta', revision: 3 }))
  b.publish(TWIN_STATE_TOPIC, twinStatePayload(B, { instanceId: 'other-twin', kind: 'delta', revision: 1 }))

  const toB = b.received(TWIN_STATE_TOPIC, twinStateFilter(B, SHARED))
  assert.deepEqual(toB.map(p => [p.domainId, p.twinState.kind, p.twinState.revision]), [[B, 'delta', 3]])

  const toA = b.received(TWIN_STATE_TOPIC, twinStateFilter(A, SHARED))
  assert.deepEqual(toA.map(p => [p.twinState.kind, p.twinState.revision]), [['snapshot', 7], ['delta', 8]])
})

test('twin-reference-progress: B receives none of A\'s import progress for a source of the same name', () => {
  const b = bus()
  b.publish(REFERENCE_PROGRESS_TOPIC, referenceProgressPayload(A, { source: SHARED, step: 'master', done: 1, error: 'A: connection refused' }))
  b.publish(REFERENCE_PROGRESS_TOPIC, referenceProgressPayload(B, { source: SHARED, step: 'master', done: 2 }))

  const toB = b.received(REFERENCE_PROGRESS_TOPIC, referenceProgressFilter(B, SHARED))
  assert.deepEqual(toB.map(p => [p.domainId, p.twinReferenceProgress.done]), [[B, 2]])
})

test('a payload cannot be built without a domain', () => {
  assert.throws(() => twinStatePayload('', { instanceId: SHARED, kind: 'delta' }), /domainId is required/)
  assert.throws(() => referenceProgressPayload('', { source: SHARED }), /domainId is required/)
})

/* ── every sender goes through the channel ────────────────────────────────── */

const SERVER = join(dirname(fileURLToPath(import.meta.url)), '..', 'server')
const CHANNELS = new Set(['engine/twin-state-channel.ts', 'service/reference/reference-progress-channel.ts'])

function serverSources(): { path: string; code: string }[] {
  const out: { path: string; code: string }[] = []
  const walk = (dir: string) => {
    for (const name of readdirSync(dir)) {
      const p = join(dir, name)
      if (statSync(p).isDirectory()) walk(p)
      else if (p.endsWith('.ts')) {
        /* Comments describe the old shape; count code only. */
        const code = readFileSync(p, 'utf8').replace(/\/\*[\s\S]*?\*\//g, '').replace(/(^|[^:])\/\/.*$/gm, '$1')
        out.push({ path: relative(SERVER, p), code })
      }
    }
  }
  walk(SERVER)
  return out
}

test('no sender builds a channel payload by hand, and nothing names the topics but the channels', () => {
  const offenders: string[] = []
  for (const { path, code } of serverSources()) {
    if (CHANNELS.has(path)) continue
    if (/\btwinState\s*:\s*\{/.test(code)) offenders.push(`${path}: builds { twinState: … } by hand`)
    if (/\btwinReferenceProgress\s*:\s*\{/.test(code)) offenders.push(`${path}: builds { twinReferenceProgress: … } by hand`)
    if (/['"`]twin-state['"`]/.test(code)) offenders.push(`${path}: names 'twin-state'`)
    if (/['"`]twin-reference-progress['"`]/.test(code)) offenders.push(`${path}: names 'twin-reference-progress'`)
  }
  assert.deepEqual(offenders, [])
})

test('the senders are really there — the scan above is not passing by finding nothing', () => {
  const callers = serverSources()
    .filter(s => !CHANNELS.has(s.path))
    .map(s => s.code)
    .join('\n')
  /* engine: runtime subscription + flush; the subscription's first snapshot. */
  assert.equal((callers.match(/twinStatePayload\(/g) ?? []).length, 3, 'twinStatePayload call sites')
  /* resync and import. */
  assert.equal((callers.match(/referenceProgressPayload\(/g) ?? []).length, 2, 'referenceProgressPayload call sites')
})