aiturk-hermes-ide/apps/desktop/electron/mobile-relay-client.test.ts

/**
 * Tests for electron/mobile-relay-client.ts — the turkservis.online tunnel
 * client (contract: aiturk-mobil-tunel-spec.md v3, §5 "IDE").
 *
 * Part 1 drives the whole path the way a phone does: a fake relay (an
 * in-process WebSocket server on 127.0.0.1 that speaks the gateway's host
 * protocol) hands out a byte stream, the test builds REAL TLS on top of that
 * stream with node:tls and pins the certificate fingerprint, and talks HTTP /
 * WebSocket to the real LAN proxy behind it. Part 2 checks the client alone
 * with a fake socket and a fake clock.
 *
 * Nothing leaves 127.0.0.1; turkservis.online is never contacted.
 */

import crypto from 'node:crypto'
import { EventEmitter } from 'node:events'
import http from 'node:http'
import net from 'node:net'
import { Duplex } from 'node:stream'
import tls from 'node:tls'

import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import { WebSocket, WebSocketServer } from 'ws'

import { fingerprintOfDer } from './mobile-lan-cert'
import { createMobileLink, type MobileLinkEvent } from './mobile-lan-proxy'
import {
  createRelayClient,
  encodeRelayFrame,
  isRelayId,
  parseRelaySession,
  registerRelaySession,
  RELAY_BACKOFF_MS,
  RELAY_BUFFER_LIMIT_BYTES,
  RELAY_CHUNK_BYTES,
  RELAY_MAX_STREAMS,
  RELAY_SILENCE_MS,
  type RelayCredentials,
  relayHostUrl,
  splitRelayChunks
} from './mobile-relay-client'

const SESSION = crypto.randomBytes(32).toString('base64url')
const RELAY_IDS = ['3f2b8c1e-5d47-4a90-8e21-6b0c9d7a1f55', '9a1d0c2b-7e63-4f18-b5a4-2c8e6f013d77']
const HOST_TOKENS = [`tsmr_${'A'.repeat(43)}`, `tsmr_${'B'.repeat(43)}`]
const BIG = crypto.randomBytes(1536 * 1024)
const sha = (data: Buffer) => crypto.createHash('sha256').update(data).digest('hex')

async function until(check: () => boolean, ms = 5000) {
  const end = Date.now() + ms

  while (!check()) {
    if (Date.now() > end) {
      throw new Error('condition not reached in time')
    }

    await new Promise(resolve => setTimeout(resolve, 10))
  }
}

// ── fake backend (what the proxy forwards to) ───────────────────────────────
let backend: http.Server
let backendPort = 0
let backendWss: WebSocketServer

beforeAll(async () => {
  backend = http.createServer((req, res) => {
    const url = new URL(req.url ?? '/', 'http://x')

    if (url.pathname === '/') {
      res.writeHead(200, { 'content-type': 'text/html' })

      return res.end(`<script>window.__HERMES_SESSION_TOKEN__="${SESSION}";</script>`)
    }

    if (url.pathname !== '/api/status' && req.headers['x-hermes-session-token'] !== SESSION) {
      res.writeHead(401, { 'content-type': 'application/json' })

      return res.end('{"detail":"Unauthorized"}')
    }

    if (url.pathname === '/api/big') {
      res.writeHead(200, { 'content-type': 'application/octet-stream', 'content-length': String(BIG.length) })

      return res.end(BIG)
    }

    if (url.pathname === '/api/echo') {
      const chunks: Buffer[] = []

      req.on('data', chunk => chunks.push(chunk))
      req.on('end', () => {
        const body = Buffer.concat(chunks)

        res.writeHead(200, { 'content-type': 'application/json' })
        res.end(JSON.stringify({ bytes: body.length, sha256: sha(body) }))
      })

      return
    }

    res.writeHead(200, { 'content-type': 'application/json' })
    res.end(JSON.stringify({ ok: true, path: url.pathname }))
  })
  backendWss = new WebSocketServer({ noServer: true })
  backend.on('upgrade', (req, socket, head) => {
    const url = new URL(req.url ?? '/', 'http://x')

    if (url.searchParams.get('token') !== SESSION) {
      return socket.end('HTTP/1.1 403 Forbidden\r\n\r\n')
    }

    backendWss.handleUpgrade(req, socket, head, ws => ws.on('message', (data, isBinary) => ws.send(data, { binary: isBinary })))
  })
  await new Promise<void>(resolve => backend.listen(0, '127.0.0.1', resolve))
  backendPort = (backend.address() as net.AddressInfo).port
})

afterAll(async () => {
  backendWss.close()
  backend.closeAllConnections()
  await new Promise(resolve => backend.close(resolve))
})

// ── fake relay: the gateway side of `WS /host/<relayId>` ────────────────────
async function startFakeRelay() {
  const tokens = new Map<string, string>()
  const wss = new WebSocketServer({ host: '127.0.0.1', port: 0, maxPayload: 1024 * 1024 })
  const streams = new Map<number, { duplex: Duplex; opened: () => void; closedByHost: null | number }>()
  /** Every data payload that crossed the relay, both directions: what turkservis.online would see. */
  const seen: Buffer[] = []
  const connections: { relayId: string; authorization: string | undefined }[] = []
  const closes: { id: number; code: number }[] = []
  let frameSizes: number[] = []
  let host: null | WebSocket = null
  let nextId = 7
  let pongs = 0

  await new Promise<void>(resolve => wss.once('listening', resolve))

  wss.on('connection', (ws, req) => {
    const relayId = String(req.url).split('/').pop() ?? ''

    connections.push({ relayId, authorization: req.headers.authorization })

    if (!tokens.has(relayId) || req.headers.authorization !== `Bearer ${tokens.get(relayId)}`) {
      return ws.close(4401)
    }

    host?.close(4409)
    host = ws
    ws.on('close', () => {
      if (host === ws) {
        host = null
      }

      for (const stream of streams.values()) {
        stream.duplex.destroy()
      }

      streams.clear()
    })
    ws.on('message', (data, isBinary) => {
      const frame = data as Buffer

      if (isBinary) {
        const stream = streams.get(frame.readUInt32BE(0))

        frameSizes.push(frame.length - 4)
        seen.push(frame.subarray(4))
        stream?.duplex.push(frame.subarray(4))

        return
      }

      const message = JSON.parse(frame.toString('utf8'))

      if (message.t === 'pong') {
        pongs++
      } else if (message.t === 'opened') {
        streams.get(message.id)?.opened()
      } else if (message.t === 'close') {
        const stream = streams.get(message.id)

        closes.push({ id: message.id, code: message.code })

        if (stream) {
          stream.closedByHost = message.code
          streams.delete(message.id)
          stream.duplex.push(null)
          stream.duplex.destroy()
        }
      }
    })
  })

  return {
    port: (wss.address() as net.AddressInfo).port,
    tokens,
    seen,
    connections,
    closes,
    get host() {
      return host
    },
    get pongs() {
      return pongs
    },
    get streamCount() {
      return streams.size
    },
    takeFrameSizes() {
      const sizes = frameSizes

      frameSizes = []

      return sizes
    },
    /** What the gateway does when a phone connects to `/t/<relayId>`: `open`, wait for `opened`, then raw bytes. */
    openStream(): Promise<Duplex> {
      return new Promise((resolve, reject) => {
        const ws = host
        const id = nextId++

        if (!ws) {
          return reject(new Error('host_offline'))
        }

        const entry = {
          closedByHost: null as null | number,
          opened: () => resolve(duplex),
          duplex: null as unknown as Duplex
        }

        const duplex = new Duplex({
          read() {},
          write(chunk: Buffer, _encoding, done) {
            for (const part of splitRelayChunks(chunk)) {
              seen.push(part)
              ws.send(encodeRelayFrame(id, part))
            }

            done()
          },
          destroy(error, done) {
            if (streams.get(id) === entry) {
              streams.delete(id)
              ws.send(JSON.stringify({ t: 'close', id, code: 1000 }))
            }

            done(error)
          }
        })

        entry.duplex = duplex
        streams.set(id, entry)
        ws.send(JSON.stringify({ t: 'open', id }))
        setTimeout(() => reject(new Error(`not opened (host closed with ${entry.closedByHost})`)), 3000)
      })
    },
    send: (value: unknown) => host?.send(typeof value === 'string' || Buffer.isBuffer(value) ? value : JSON.stringify(value)),
    stop: () =>
      new Promise<void>(resolve => {
        for (const client of wss.clients) {
          client.terminate()
        }

        wss.close(() => resolve())
      })
  }
}

type FakeRelay = Awaited<ReturnType<typeof startFakeRelay>>

/** The phone: TLS inside the tunnel stream, trusting ONLY the pinned fingerprint. */
async function phoneTls(relay: FakeRelay, cert: string, fingerprint: string): Promise<tls.TLSSocket> {
  const stream = await relay.openStream()

  return new Promise((resolve, reject) => {
    const socket = tls.connect({
      socket: stream,
      ca: cert,
      checkServerIdentity: (_host, peer) => (fingerprintOfDer(peer.raw) === fingerprint ? undefined : new Error('tls:pin'))
    })

    socket.once('secureConnect', () => resolve(socket))
    socket.once('error', reject)
  })
}

interface PhoneCall {
  token?: string
  body?: Buffer | string
  headers?: Record<string, string>
}

async function phoneCall(relay: FakeRelay, pin: { cert: string; fingerprint: string }, method: string, path: string, options: PhoneCall = {}) {
  const socket = await phoneTls(relay, pin.cert, pin.fingerprint)

  return new Promise<{ status: number; body: Buffer; json: any }>((resolve, reject) => {
    const req = http.request(
      {
        // No agent: Node then uses exactly this socket (the TLS session inside the tunnel).
        createConnection: () => socket,
        host: '127.0.0.1',
        port: 9, // never dialled: the TLS socket inside the tunnel is the connection
        method,
        path,
        headers: {
          ...(options.token ? { authorization: `Bearer ${options.token}` } : {}),
          ...(options.body ? { 'content-length': String(Buffer.byteLength(options.body)) } : {}),
          ...options.headers
        }
      },
      res => {
        const chunks: Buffer[] = []

        res.on('data', chunk => chunks.push(chunk))
        res.on('end', () => {
          const body = Buffer.concat(chunks)
          let json: any = null

          try {
            json = JSON.parse(body.toString('utf8'))
          } catch {
            json = null
          }

          socket.destroy()
          resolve({ status: res.statusCode ?? 0, body, json })
        })
      }
    )

    req.on('error', error => reject(new Error(`${method} ${path}: ${error.message}`)))
    req.end(options.body)
  })
}

// ── controller wired to the fake relay ──────────────────────────────────────
function tunnelHarness(relay: FakeRelay) {
  const vault = { value: {} as Record<string, any> }
  const store = { value: null as any }
  const events: MobileLinkEvent[] = []
  const logs: string[] = []
  const firewallCalls: string[] = []
  const dialed: string[] = []
  let registered = 0

  const link = createMobileLink({
    vault: {
      load: async () => structuredClone(vault.value),
      save: async (value: Record<string, any>) => {
        vault.value = structuredClone(value)
      }
    },
    store: {
      load: async () => structuredClone(store.value),
      save: async (value: any) => {
        store.value = structuredClone(value)
      }
    },
    resolveBackend: async () => ({ baseUrl: `http://127.0.0.1:${backendPort}`, token: SESSION }),
    isRemoteMode: () => false,
    firewall: {
      addRule: async (port: number) => {
        firewallCalls.push(`add:${port}`)

        return { ok: true }
      },
      removeRule: async () => {
        firewallCalls.push('remove')

        return { ok: true }
      },
      queryRule: async () => ({ state: 'absent' })
    },
    fetch: async (url: string) => {
      if (url.endsWith('/mobile-relay/sessions')) {
        const index = registered++

        relay.tokens.set(RELAY_IDS[index], HOST_TOKENS[index])

        return { status: 201, json: async () => ({ relayId: RELAY_IDS[index], hostToken: HOST_TOKENS[index] }) }
      }

      return { status: 201, json: async () => ({ offerId: '0f8fad5b-d9cb-469f-a165-70867728950e', cancelToken: `tsmp_${'c'.repeat(43)}` }) }
    },
    // The real `ws` client with the real header; only the origin is swapped for the fake relay.
    relayConnect: (url: string, headers: Record<string, string>) => {
      dialed.push(url)

      return new WebSocket(url.replace('wss://turkservis.online/v1/aiturk/mobile-relay', `ws://127.0.0.1:${relay.port}`), { headers, perMessageDeflate: false })
    },
    addresses: async () => ['192.168.1.20'],
    version: '1.0.0-test',
    pcName: 'TEST-PC',
    platform: 'win32',
    listenHost: '127.0.0.1',
    ports: [0],
    log: (line: string) => logs.push(line)
  } as any)

  link.onEvent(event => events.push(event))

  return { link, vault, store, events, logs, firewallCalls, dialed }
}

describe('tunnel: phone → fake relay → relay client → real TLS proxy', () => {
  it('real end-to-end TLS through the tunnel: hello, 401, 404, 403, claim, status, WebSocket; the relay sees ciphertext only', async () => {
    const relay = await startFakeRelay()
    const h = tunnelHarness(relay)

    await h.link.enable()
    expect(h.firewallCalls).toEqual([])

    const state = await h.link.pairBegin()
    const pin = { cert: h.vault.value.lanTls.cert as string, fingerprint: h.vault.value.lanTls.fingerprint as string }

    expect(state.pairing!.qr).toBe(`aiturk://pair?v=1&fp=${pin.fingerprint}&c=${state.pairing!.code}&r=${RELAY_IDS[0]}`)
    expect(isRelayId(new URLSearchParams(state.pairing!.qr.split('?')[1]).get('r'))).toBe(true)
    await until(() => relay.host !== null && h.events.some(event => event.state.relay.status === 'connected'))
    expect(h.dialed).toEqual([`wss://turkservis.online/v1/aiturk/mobile-relay/host/${RELAY_IDS[0]}`])
    expect(relay.connections).toEqual([{ relayId: RELAY_IDS[0], authorization: `Bearer ${HOST_TOKENS[0]}` }])
    expect(await h.link.state()).toMatchObject({ listening: true, relay: { status: 'connected', ready: true }, lan: { direct: false, listening: false } })

    // The certificate the phone sees through the tunnel is the proxy's own: nobody terminated TLS on the way.
    const probe = await phoneTls(relay, pin.cert, pin.fingerprint)

    expect(fingerprintOfDer(probe.getPeerCertificate().raw)).toBe(pin.fingerprint)
    probe.destroy()
    await expect(phoneTls(relay, pin.cert, 'x'.repeat(43))).rejects.toThrow('tls:pin')

    const hello = await phoneCall(relay, pin, 'GET', '/aiturk/pair/hello')

    expect(hello.status).toBe(200)
    expect(hello.json).toEqual({ app: 'aiturk-ide', version: '1.0.0-test', pairing: true })

    // The proxy's rules apply unchanged: the tunnel is not a way around them.
    expect((await phoneCall(relay, pin, 'GET', '/api/sessions')).status).toBe(401)
    expect((await phoneCall(relay, pin, 'GET', '/api/sessions', { token: SESSION })).status).toBe(401)

    const root = await phoneCall(relay, pin, 'GET', '/')

    expect(root.status).toBe(404)
    expect(root.body.toString()).not.toContain(SESSION)

    const claim = await phoneCall(relay, pin, 'POST', '/aiturk/pair/claim', {
      body: JSON.stringify({ code: state.pairing!.code, device_name: 'Pixel 9' }),
      headers: { 'content-type': 'application/json' }
    })

    expect(claim.status).toBe(200)
    expect(claim.json.token).toMatch(/^amd_[A-Za-z0-9_-]{43}$/)
    await until(() => h.events.some(event => event.type === 'paired'))

    const token = claim.json.token as string
    const status = await phoneCall(relay, pin, 'GET', '/api/status', { token })

    expect(status.status).toBe(200)
    expect(status.json).toEqual({ ok: true, path: '/api/status' })
    expect((await phoneCall(relay, pin, 'GET', '/api/env', { token })).status).toBe(403)
    expect((await phoneCall(relay, pin, 'POST', '/api/config', { token, body: '{}' })).status).toBe(403)

    // ≥ 1 MiB each way, byte for byte, in frames of at most 64 KiB.
    relay.takeFrameSizes()

    const big = await phoneCall(relay, pin, 'GET', '/api/big', { token })

    expect(big.status).toBe(200)
    expect(big.body.length).toBe(BIG.length)
    expect(sha(big.body)).toBe(sha(BIG))

    const sizes = relay.takeFrameSizes()

    expect(sizes.reduce((sum, size) => sum + size, 0)).toBeGreaterThan(BIG.length)
    expect(Math.max(...sizes)).toBeLessThanOrEqual(RELAY_CHUNK_BYTES)

    const upload = await phoneCall(relay, pin, 'POST', '/api/echo', { token, body: BIG, headers: { 'content-type': 'application/octet-stream' } })

    expect(upload.json).toEqual({ bytes: BIG.length, sha256: sha(BIG) })

    // WebSocket (101) with the device token in the Authorization header; text and binary both ways; close.
    const wsSocket = await phoneTls(relay, pin.cert, pin.fingerprint)
    const ws = new WebSocket('ws://127.0.0.1:9/api/ws', { createConnection: () => wsSocket, headers: { authorization: `Bearer ${token}` } } as any)
    const got: { data: Buffer; binary: boolean }[] = []

    ws.on('message', (data, binary) => got.push({ data: data as Buffer, binary }))
    await new Promise((resolve, reject) => {
      ws.once('open', resolve)
      ws.once('error', reject)
      ws.once('unexpected-response', (_req, res) => reject(new Error(`ws ${res.statusCode}`)))
    })
    ws.send('merhaba telefon')
    ws.send(Buffer.from([0, 1, 2, 254, 255]))
    await until(() => got.length === 2)
    expect(got[0]).toEqual({ data: Buffer.from('merhaba telefon'), binary: false })
    expect(got[1]).toEqual({ data: Buffer.from([0, 1, 2, 254, 255]), binary: true })

    const closed = new Promise<number>(resolve => ws.once('close', code => resolve(code)))

    ws.close(1000)
    expect(await closed).toBe(1000)

    // Forbidden WebSocket path and a missing token are refused by the proxy, through the tunnel too.
    for (const [path, headers, expected] of [
      ['/api/pty', { authorization: `Bearer ${token}` }, 403],
      ['/api/ws', {}, 401]
    ] as const) {
      const refusedSocket = await phoneTls(relay, pin.cert, pin.fingerprint)
      const refused = new WebSocket(`ws://127.0.0.1:9${path}`, { createConnection: () => refusedSocket, headers } as any)

      const outcome = await new Promise<number>(resolve => {
        refused.once('unexpected-response', (_req, res) => resolve(res.statusCode ?? 0))
        refused.once('open', () => resolve(101))
        refused.once('error', () => resolve(0))
      })

      expect(outcome).toBe(expected)
      refusedSocket.destroy()
    }

    // What the relay carried: TLS records. No token, no code, no path, no HTTP, no plaintext payload.
    const wire = Buffer.concat(relay.seen)

    expect(wire.length).toBeGreaterThan(2 * BIG.length)

    for (const secret of [token, token.slice(4), state.pairing!.code, SESSION, 'HTTP/1.1', '/api/status', 'aiturk/pair', 'merhaba telefon', 'authorization']) {
      expect(wire.includes(Buffer.from(secret))).toBe(false)
    }

    expect(wire.includes(BIG.subarray(0, 64))).toBe(false)

    // Streams are released when the phone side ends.
    await until(() => relay.streamCount === 0)

    // Revoking the phone in the IDE kills its token on the tunnel path as well; with no phone left the relay socket closes.
    await h.link.deviceRevoke(claim.json.device_id)
    await until(() => relay.host === null)
    expect((await h.link.state()).relay.status).toBe('off')
    await expect(relay.openStream()).rejects.toThrow('host_offline')

    // The host token stayed in the vault and the Authorization header.
    const exposed = [h.logs.join('\n'), JSON.stringify(h.events), JSON.stringify(await h.link.state()), JSON.stringify(h.store.value)].join('\n')

    expect(exposed).not.toContain('tsmr_')
    expect(exposed).not.toContain(token)
    expect(h.vault.value.relay.hostToken).toBe(HOST_TOKENS[0])
    await h.link.stop()
    await relay.stop()
  }, 30_000)

  it('a dropped relay socket ends the phone streams and the client comes back; a stream with no listener is refused with 4503', async () => {
    const relay = await startFakeRelay()
    const timers: { fn: () => void; ms: number }[] = []
    const states: string[] = []
    let port: null | number = null
    const server = net.createServer(socket => socket.on('data', data => socket.write(data)).on('error', () => undefined))

    await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
    relay.tokens.set(RELAY_IDS[0], HOST_TOKENS[0])

    const client = createRelayClient({
      connect: (url, headers) => new WebSocket(url.replace('wss://turkservis.online/v1/aiturk/mobile-relay', `ws://127.0.0.1:${relay.port}`), { headers }),
      fetch: async () => ({ status: 500, json: async () => ({}) }),
      dial: () => (port === null ? null : net.connect(port, '127.0.0.1')),
      credentials: { load: async () => ({ relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] }), save: async () => undefined, clear: async () => undefined },
      // Only the reconnect timer is faked; the 75 s silence watchdog is simply never fired.
      setTimer: (fn, ms) => {
        timers.push({ fn, ms })

        return fn
      },
      clearTimer: handle => {
        const index = timers.findIndex(timer => timer.fn === handle)

        if (index >= 0) {
          timers.splice(index, 1)
        }
      },
      onState: state => states.push(`${state.status}${state.code ? `:${state.code}` : ''}`)
    })

    client.start()
    await until(() => relay.host !== null && client.state().status === 'connected')

    // No listener behind the client: the stream is closed with 4503 instead of hanging.
    await expect(relay.openStream()).rejects.toThrow('4503')

    port = (server.address() as net.AddressInfo).port

    const stream = await relay.openStream()
    const echoed: Buffer[] = []

    stream.on('data', chunk => echoed.push(chunk))
    stream.write(Buffer.from('ping-1'))
    await until(() => Buffer.concat(echoed).toString() === 'ping-1')
    expect(client.streams()).toBe(1)

    relay.send({ t: 'ping' })
    await until(() => relay.pongs === 1)

    // The gateway goes away.
    const ended = new Promise(resolve => stream.once('close', resolve))

    relay.host!.terminate()
    await ended
    await until(() => client.state().status === 'failed')
    expect(client.streams()).toBe(0)
    expect(states.at(-1)).toBe('failed:connect_failed')
    expect(timers.filter(timer => timer.ms === RELAY_BACKOFF_MS[0])).toHaveLength(1)

    timers.find(timer => timer.ms === RELAY_BACKOFF_MS[0])!.fn()
    await until(() => relay.host !== null && client.state().status === 'connected')

    const again = await relay.openStream()

    again.write(Buffer.from('ping-2'))
    await new Promise<void>(resolve => again.once('data', () => resolve()))

    // The gateway closes one stream: the local connection goes with it, the others stay.
    again.destroy()
    await until(() => client.streams() === 0)

    client.stop()
    await until(() => relay.host === null)
    expect(client.state()).toEqual({ status: 'off', code: null })
    await relay.stop()
    await new Promise(resolve => server.close(resolve))
  }, 20_000)
})

// ── the client alone: fake socket, fake clock ───────────────────────────────
class FakeSocket extends EventEmitter {
  sent: (Buffer | string)[] = []
  closed: null | number = null
  terminated = false
  bufferedAmount = 0

  constructor(
    public url: string,
    public headers: Record<string, string>
  ) {
    super()
  }

  send(data: Buffer | string) {
    this.sent.push(data)
  }

  close(code = 1000) {
    this.closed = code
  }

  terminate() {
    this.terminated = true
  }

  control() {
    return this.sent.filter((item): item is string => typeof item === 'string').map(item => JSON.parse(item))
  }

  binary() {
    return this.sent.filter((item): item is Buffer => Buffer.isBuffer(item))
  }
}

class FakeLocal extends EventEmitter {
  written: Buffer[] = []
  destroyed = false
  paused = 0
  resumed = 0
  writableLength = 0

  write(data: Buffer) {
    this.written.push(data)

    return true
  }

  destroy() {
    if (!this.destroyed) {
      this.destroyed = true
      this.emit('close')
    }
  }

  pause() {
    this.paused++
  }

  resume() {
    this.resumed++
  }
}

function clientHarness(over: { stored?: null | RelayCredentials; sessions?: (number | 'throw')[] } = {}) {
  const sockets: FakeSocket[] = []
  const locals: FakeLocal[] = []
  const timers: { fn: () => void; ms: number; id: number }[] = []
  const logs: string[] = []
  const states: string[] = []
  const saved: { credentials: RelayCredentials; reissued: boolean }[] = []
  const fetches: { url: string; init: any }[] = []
  const sessions = over.sessions ?? []
  const box = { stored: over.stored === undefined ? null : over.stored, cleared: 0, dial: true, registered: 0, timerId: 0 }

  const client = createRelayClient({
    connect: (url, headers) => {
      const socket = new FakeSocket(url, headers)

      sockets.push(socket)

      return socket
    },
    fetch: async (url, init) => {
      fetches.push({ url, init })

      const outcome = sessions.shift() ?? 201

      if (outcome === 'throw') {
        throw new Error(`network down ${HOST_TOKENS[0]}`)
      }

      const index = box.registered

      if (outcome === 201) {
        box.registered++
      }

      return { status: outcome, json: async () => ({ relayId: RELAY_IDS[index], hostToken: HOST_TOKENS[index] }) }
    },
    dial: () => {
      if (!box.dial) {
        return null
      }

      const local = new FakeLocal()

      locals.push(local)

      return local
    },
    credentials: {
      load: async () => box.stored,
      save: async (credentials, info) => {
        box.stored = credentials
        saved.push({ credentials, reissued: info.reissued })
      },
      clear: async () => {
        box.stored = null
        box.cleared++
      }
    },
    setTimer: (fn, ms) => {
      const timer = { fn, ms, id: ++box.timerId }

      timers.push(timer)

      return timer
    },
    clearTimer: handle => {
      const index = timers.indexOf(handle as (typeof timers)[number])

      if (index >= 0) {
        timers.splice(index, 1)
      }
    },
    log: line => logs.push(line),
    onState: state => states.push(`${state.status}${state.code ? `:${state.code}` : ''}`)
  })

  /** Fire the pending timer with this delay (the fake clock). */
  const fire = (ms: number) => {
    const index = timers.findIndex(timer => timer.ms === ms)

    if (index < 0) {
      throw new Error(`no timer with ${ms} ms; pending: ${timers.map(timer => timer.ms).join(',')}`)
    }

    timers.splice(index, 1)[0].fn()
  }

  const settle = () => new Promise(resolve => setTimeout(resolve, 0))
  const retryDelays = () => timers.filter(timer => timer.ms !== RELAY_SILENCE_MS && timer.ms >= 1000).map(timer => timer.ms)

  return { client, sockets, locals, timers, logs, states, saved, fetches, box, fire, settle, retryDelays }
}

describe('relay client', () => {
  it('registers once, stores the registration, sends the token only in the Authorization header', async () => {
    const h = clientHarness()

    expect(h.client.state()).toEqual({ status: 'off', code: null })
    h.client.start()
    await h.settle()

    expect(h.fetches).toHaveLength(1)
    expect(h.fetches[0].url).toBe('https://turkservis.online/v1/aiturk/mobile-relay/sessions')
    expect(h.fetches[0].init).toMatchObject({ method: 'POST', body: '{}', redirect: 'error' })
    expect(h.fetches[0].init.signal).toBeInstanceOf(AbortSignal)
    expect(h.saved).toEqual([{ credentials: { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] }, reissued: false }])
    expect(h.sockets[0].url).toBe(relayHostUrl(RELAY_IDS[0]))
    expect(h.sockets[0].url).toBe(`wss://turkservis.online/v1/aiturk/mobile-relay/host/${RELAY_IDS[0]}`)
    expect(h.sockets[0].headers).toEqual({ Authorization: `Bearer ${HOST_TOKENS[0]}` })
    expect(h.client.state().status).toBe('connecting')

    h.sockets[0].emit('open')
    expect(h.client.state()).toEqual({ status: 'connected', code: null })

    h.sockets[0].emit('message', Buffer.from('{"t":"ping"}'), false)
    expect(h.sockets[0].control()).toEqual([{ t: 'pong' }])

    // A stored registration is reused: the phone's saved relay id must not change between runs.
    const second = clientHarness({ stored: { relayId: RELAY_IDS[1], hostToken: HOST_TOKENS[1] } })

    second.client.start()
    await second.settle()
    expect(second.fetches).toHaveLength(0)
    expect(second.sockets[0].headers).toEqual({ Authorization: `Bearer ${HOST_TOKENS[1]}` })
    expect(await second.client.ensureRegistered()).toBe(RELAY_IDS[1])

    expect(h.logs).toEqual(['relay connecting', 'relay connected'])
    expect(h.logs.join('\n') + JSON.stringify(h.client.state()) + h.states.join('\n')).not.toContain('tsmr_')
  })

  it('reconnects after 2 s, then with growing delays up to 30 s; a successful open starts over at 2 s', async () => {
    const h = clientHarness({ stored: { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] } })

    h.client.start()
    await h.settle()
    h.sockets[0].emit('open')
    h.sockets[0].emit('close', 1006)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'connect_failed' })

    const delays: number[] = []

    for (let round = 0; round < 7; round++) {
      const [delay] = h.retryDelays()

      delays.push(delay)
      h.fire(delay)
      await h.settle()
      expect(h.client.state().status).toBe('connecting')
      // Connection refused: `ws` reports an error and then the close.
      h.sockets.at(-1)!.emit('error', new Error('ECONNREFUSED'))
      h.sockets.at(-1)!.emit('close', 1006)
      expect(h.retryDelays()).toHaveLength(1)
    }

    expect(delays).toEqual([2000, 4000, 8000, 16_000, 30_000, 30_000, 30_000])
    expect(RELAY_BACKOFF_MS).toEqual([2000, 4000, 8000, 16_000, 30_000])

    h.fire(30_000)
    await h.settle()
    h.sockets.at(-1)!.emit('open')
    expect(h.client.state().status).toBe('connected')
    h.sockets.at(-1)!.emit('close', 1001)
    expect(h.retryDelays()).toEqual([2000])

    // A silent gateway (no ping for 75 s) is treated as gone.
    h.fire(2000)
    await h.settle()
    h.sockets.at(-1)!.emit('open')
    h.fire(RELAY_SILENCE_MS)
    expect(h.sockets.at(-1)!.terminated).toBe(true)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'timeout' })

    // stop() cancels the pending retry.
    h.client.stop()
    expect(h.timers).toEqual([])
    expect(h.client.state()).toEqual({ status: 'off', code: null })
    expect(h.fetches).toHaveLength(0)
  })

  it('4401 replaces the registration and says so; 4409 waits 30 s as "replaced"; 4429 is "busy" with the normal delay', async () => {
    const h = clientHarness({ stored: { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] } })

    h.client.start()
    await h.settle()
    h.sockets[0].emit('open')
    h.sockets[0].emit('close', 4401)
    await h.settle()

    expect(h.box.cleared).toBe(1)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'token_rejected' })
    expect(h.fetches).toHaveLength(0)
    expect(h.retryDelays()).toEqual([2000])

    h.fire(2000)
    await h.settle()
    // A NEW registration; the controller learns that the phones' stored id is dead.
    expect(h.fetches).toHaveLength(1)
    expect(h.saved).toEqual([{ credentials: { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] }, reissued: true }])
    expect(h.sockets).toHaveLength(2)
    h.sockets[1].emit('open')
    expect(h.client.state().status).toBe('connected')

    // Another IDE instance took the registration over: no reconnect loop.
    h.sockets[1].emit('close', 4409)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'replaced' })
    expect(h.retryDelays()).toEqual([30_000])
    expect(h.box.cleared).toBe(1)

    h.fire(30_000)
    await h.settle()
    h.sockets[2].emit('close', 4429)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'busy' })
    expect(h.retryDelays()).toEqual([4000])

    // A refusal before the WebSocket exists (HTTP 401 on the upgrade) counts as 4401 too.
    h.fire(4000)
    await h.settle()

    const destroyed: string[] = []

    h.sockets[3].emit('unexpected-response', { destroy: () => destroyed.push('req') }, { statusCode: 401, resume: () => destroyed.push('res') })
    await h.settle()
    expect(destroyed).toEqual(['res', 'req'])
    expect(h.box.cleared).toBe(2)
    expect(h.client.state()).toEqual({ status: 'failed', code: 'token_rejected' })
    expect(h.states).toContain('failed:token_rejected')
    expect(h.logs.join('\n')).not.toContain('tsmr_')
    h.client.stop()
  })

  it('a failing registration is a fixed code (no upstream text) and is retried; renew() swaps the registration only when the new one is stored', async () => {
    const h = clientHarness({ sessions: ['throw', 429, 500, 201, 500, 201] })

    h.client.start()
    await h.settle()
    expect(h.client.state()).toEqual({ status: 'failed', code: 'register_failed' })
    expect(h.sockets).toHaveLength(0)
    h.fire(2000)
    await h.settle()
    expect(h.client.state()).toEqual({ status: 'failed', code: 'rate_limited' })
    h.fire(4000)
    await h.settle()
    expect(h.client.state()).toEqual({ status: 'failed', code: 'register_failed' })
    expect(await h.client.ensureRegistered()).toBe(RELAY_IDS[0])
    h.fire(8000)
    await h.settle()
    h.sockets[0].emit('open')
    expect(h.client.state().status).toBe('connected')

    expect(await h.client.renew()).toBe(false)
    expect(h.box.stored).toEqual({ relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] })
    expect(h.sockets[0].closed).toBeNull()

    expect(await h.client.renew()).toBe(true)
    expect(h.saved.at(-1)).toEqual({ credentials: { relayId: RELAY_IDS[1], hostToken: HOST_TOKENS[1] }, reissued: true })
    expect(h.sockets[0].closed).toBe(1000)
    await h.settle()
    expect(h.sockets[1].url).toBe(relayHostUrl(RELAY_IDS[1]))
    expect(h.sockets[1].headers).toEqual({ Authorization: `Bearer ${HOST_TOKENS[1]}` })

    expect(h.logs.join('\n') + h.states.join('\n')).not.toContain('tsmr_')
    expect(h.logs.every(line => /^relay (?:off|connecting|connected|failed:[a-z_]+)$/.test(line))).toBe(true)
    h.client.stop()

    const broken = createRelayClient({
      connect: () => {
        throw new Error('must not connect')
      },
      fetch: async () => ({ status: 201, json: async () => ({}) }),
      dial: () => null,
      credentials: {
        load: async () => {
          throw new Error(`vault ${HOST_TOKENS[0]}`)
        },
        save: async () => undefined,
        clear: async () => undefined
      },
      setTimer: () => 0,
      clearTimer: () => undefined
    })

    broken.start()
    await h.settle()
    expect(broken.state()).toEqual({ status: 'failed', code: 'vault_unavailable' })
    expect(await broken.ensureRegistered()).toBeNull()
    broken.stop()
  })

  it('streams: opened / close, bytes copied untouched in ≤ 64 KiB frames, unknown ids ignored, stream and buffer limits', async () => {
    const h = clientHarness({ stored: { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] } })

    h.client.start()
    await h.settle()

    const ws = h.sockets[0]

    ws.emit('open')
    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 4_000_000_000 })), false)
    expect(h.locals).toHaveLength(1)
    // `opened` only once the TCP connection to the proxy exists.
    expect(ws.control()).toEqual([])
    h.locals[0].emit('connect')
    expect(ws.control()).toEqual([{ t: 'opened', id: 4_000_000_000 }])

    // gateway → local: payload after the 4-byte id, unchanged.
    const inbound = crypto.randomBytes(1000)

    ws.emit('message', encodeRelayFrame(4_000_000_000, inbound), true)
    ws.emit('message', encodeRelayFrame(5, Buffer.from('late')), true)
    ws.emit('message', Buffer.from([0, 0, 0]), true)
    expect(Buffer.concat(h.locals[0].written).equals(inbound)).toBe(true)

    // local → gateway: 200 KiB becomes 64 + 64 + 64 + 8 KiB, each with the stream id in front.
    const outbound = crypto.randomBytes(200 * 1024)

    h.locals[0].emit('data', outbound)

    const frames = ws.binary()

    expect(frames.map(frame => frame.length - 4)).toEqual([65_536, 65_536, 65_536, 8192])
    expect(frames.every(frame => frame.readUInt32BE(0) === 4_000_000_000)).toBe(true)
    expect(Buffer.concat(frames.map(frame => frame.subarray(4))).equals(outbound)).toBe(true)
    expect(splitRelayChunks(Buffer.alloc(0))).toEqual([])

    // Back-pressure, local → gateway: the proxy socket is paused while the host socket is behind …
    expect(h.locals[0].paused).toBe(0)
    ws.bufferedAmount = RELAY_BUFFER_LIMIT_BYTES + 1
    h.locals[0].emit('data', Buffer.from('x'))
    expect(h.locals[0].paused).toBe(1)
    h.locals[0].emit('data', Buffer.from('y'))
    expect(h.locals[0].paused).toBe(1)
    h.fire(25)
    expect(h.locals[0].resumed).toBe(0)
    // … and resumed once it has drained.
    ws.bufferedAmount = 1000
    h.fire(25)
    expect(h.locals[0].resumed).toBe(1)

    // Back-pressure, gateway → local: a proxy that does not read costs at most 1 MiB, then the stream is closed (4408).
    h.locals[0].writableLength = RELAY_BUFFER_LIMIT_BYTES + 1
    ws.emit('message', encodeRelayFrame(4_000_000_000, Buffer.from('z')), true)
    expect(h.locals[0].destroyed).toBe(true)
    expect(ws.control().at(-1)).toEqual({ t: 'close', id: 4_000_000_000, code: 4408 })
    expect(h.client.streams()).toBe(0)

    // With a host socket that can be paused (the real one), reading stops at 256 KiB and continues on drain.
    const pausable = Object.assign(ws, { pauses: 0, resumes: 0, pause: () => pausable.pauses++, resume: () => pausable.resumes++ })

    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 50 })), false)

    const slow = h.locals.at(-1)!

    slow.writableLength = 300 * 1024
    ws.emit('message', encodeRelayFrame(50, Buffer.from('a')), true)
    ws.emit('message', encodeRelayFrame(50, Buffer.from('b')), true)
    expect([pausable.pauses, pausable.resumes, slow.destroyed]).toEqual([1, 0, false])
    slow.writableLength = 0
    slow.emit('drain')
    expect([pausable.pauses, pausable.resumes]).toEqual([1, 1])
    // A stalled stream that ends must not leave the host socket paused.
    slow.writableLength = 300 * 1024
    ws.emit('message', encodeRelayFrame(50, Buffer.from('c')), true)
    slow.destroy()
    expect([pausable.pauses, pausable.resumes]).toEqual([2, 2])
    expect(ws.control().at(-1)).toEqual({ t: 'close', id: 50, code: 1000 })

    // The gateway closes a stream: local connection destroyed, no echo of the close.
    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 1 })), false)

    const first = h.locals.length - 1

    h.locals[first].emit('connect')

    const before = ws.control().length

    ws.emit('message', Buffer.from(JSON.stringify({ t: 'close', id: 1, code: 1000 })), false)
    expect(h.locals[first].destroyed).toBe(true)
    expect(ws.control()).toHaveLength(before)

    // The proxy closes the connection: the gateway is told once (1000 clean, 1011 after an error).
    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 2 })), false)
    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 3 })), false)
    h.locals[first + 1].emit('close')
    h.locals[first + 2].emit('error', new Error('ECONNRESET'))
    h.locals[first + 2].emit('close')
    expect(ws.control().slice(-2)).toEqual([
      { t: 'close', id: 2, code: 1000 },
      { t: 'close', id: 3, code: 1011 }
    ])

    // Malformed control frames and ids are ignored.
    for (const bad of ['not json', '{"t":"open"}', '{"t":"open","id":-1}', '{"t":"open","id":"7"}', '{"t":"open","id":1.5}', 'null']) {
      ws.emit('message', Buffer.from(bad), false)
    }

    expect(h.client.streams()).toBe(0)

    // At most 16 streams; the 17th is refused with 4429. Nothing listening → 4503.
    for (let id = 100; id < 100 + RELAY_MAX_STREAMS; id++) {
      ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id })), false)
    }

    expect(h.client.streams()).toBe(RELAY_MAX_STREAMS)
    ws.emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 999 })), false)
    expect(ws.control().at(-1)).toEqual({ t: 'close', id: 999, code: 4429 })

    // The relay socket drops: every local connection is closed.
    const open = h.locals.filter(local => !local.destroyed)

    expect(open).toHaveLength(RELAY_MAX_STREAMS)
    ws.emit('close', 1006)
    expect(open.every(local => local.destroyed)).toBe(true)
    expect(h.client.streams()).toBe(0)

    h.fire(2000)
    await h.settle()
    h.sockets[1].emit('open')
    h.box.dial = false
    h.sockets[1].emit('message', Buffer.from(JSON.stringify({ t: 'open', id: 1 })), false)
    expect(h.sockets[1].control()).toEqual([{ t: 'close', id: 1, code: 4503 }])
    h.client.stop()
  })

  it('session answer and relay id are validated strictly', async () => {
    const good = { relayId: RELAY_IDS[0], hostToken: HOST_TOKENS[0] }

    expect(parseRelaySession(good)).toEqual(good)
    expect(parseRelaySession({ ...good, relayId: RELAY_IDS[0].toUpperCase(), hostUrl: 'wss://evil.example/x' })).toEqual(good)

    for (const bad of [null, 'x', {}, { ...good, relayId: 'x' }, { ...good, hostToken: 'tsmr_short' }, { ...good, hostToken: `amd_${'A'.repeat(43)}` }, { relayId: 5, hostToken: 6 }]) {
      expect(parseRelaySession(bad)).toBeNull()
    }

    expect(isRelayId(RELAY_IDS[0])).toBe(true)

    for (const bad of [RELAY_IDS[0].toUpperCase(), `${RELAY_IDS[0]}/x`, `https://turkservis.online/${RELAY_IDS[0]}`, '', null, 5]) {
      expect(isRelayId(bad)).toBe(false)
    }

    expect(await registerRelaySession(async () => ({ status: 201, json: async () => good }))).toEqual(good)
    expect(await registerRelaySession(async () => ({ status: 200, json: async () => good }))).toEqual({ error: 'register_failed' })
    expect(await registerRelaySession(async () => ({ status: 429, json: async () => ({}) }))).toEqual({ error: 'rate_limited' })
    expect(await registerRelaySession(async () => ({ status: 201, json: async () => ({ relayId: 'x' }) }))).toEqual({ error: 'register_failed' })
  })
})