/// <reference types="vite/client" />
import { expect, test, beforeEach } from "vitest"
import { AMQPSession } from "../src/amqp-session.js"
import { AMQPQueue } from "../src/amqp-queue.js"
import { AMQPExchange } from "../src/amqp-exchange.js"
import { AMQPMessage } from "../src/amqp-message.js"

const WS_URL = import.meta.env.VITE_WS_URL || "ws://127.0.0.1:15670/ws/amqp"

beforeEach(() => {
  expect.hasAssertions()
})

test("AMQPSession.connect() returns a session over WebSocket", async () => {
  const session = await AMQPSession.connect(WS_URL)
  expect(session).toBeInstanceOf(AMQPSession)
  await session.stop()
})

test("session.queue() declares a queue and returns AMQPQueue", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-sq-" + Math.random(), { durable: false, autoDelete: true })
  expect(q).toBeInstanceOf(AMQPQueue)
  expect(q.name).toMatch(/^test-ws-sq-/)
  await session.stop()
})

test("AMQPQueue.publish() and get() round-trip", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-rtt-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("round-trip")
  const msg = await q.get({ noAck: true })
  expect(msg?.bodyString()).toBe("round-trip")

  await session.stop()
})

test("AMQPQueue.subscribe() delivers messages via callback", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-sub-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("hello ws queue")
  const received = await new Promise<string>((resolve) => {
    q.subscribe({ noAck: true }, (msg) => resolve(msg.bodyString()!))
  })

  expect(received).toBe("hello ws queue")
  await session.stop()
})

test("AMQPQueue.subscribe() nack", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-nack-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("nack me")
  const msg = await new Promise<AMQPMessage>((resolve) => {
    q.subscribe({ noAck: false }, (m) => {
      m.nack()
      resolve(m)
    })
  })

  expect(msg.bodyString()).toBe("nack me")
  await session.stop()
})

test("AMQPQueue.subscribe() async generator", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-gen-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("msg1")
  await q.publish("msg2")

  const received: string[] = []
  const sub = await q.subscribe({ noAck: true })
  for await (const msg of sub) {
    received.push(msg.bodyString()!)
    if (received.length >= 2) break
  }

  expect(received).toEqual(["msg1", "msg2"])
  await session.stop()
})

test("AMQPQueue.publish({ confirm: false }) sends without waiting for confirm", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-paf-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("fire and forget", { confirm: false })
  await new Promise((resolve) => setTimeout(resolve, 50))

  const msg = await q.get({ noAck: true })
  expect(msg?.bodyString()).toBe("fire and forget")
  await session.stop()
})

test("AMQPQueue.bind() and unbind()", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-bind-" + Math.random(), { durable: false, autoDelete: true })

  await expect(q.bind("amq.topic", "test.key")).resolves.toBeInstanceOf(AMQPQueue)
  await expect(q.unbind("amq.topic", "test.key")).resolves.toBeInstanceOf(AMQPQueue)

  await session.stop()
})

test("AMQPQueue.purge() empties the queue", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-purge-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("msg1", { confirm: false })
  await q.publish("msg2", { confirm: false })
  await new Promise((resolve) => setTimeout(resolve, 50))

  const result = await q.purge()
  expect(result.messageCount).toBe(2)
  await session.stop()
})

test("AMQPQueue.delete() removes the queue", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-del-" + Math.random(), { durable: false, autoDelete: false })

  await expect(q.delete()).resolves.toBeDefined()
  await session.stop()
})

test("AMQPQueue.subscribe() with unconfirmed publish round-trip", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-sessub-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("via subscribe", { confirm: false })
  const received = await new Promise<string>((resolve) => {
    void q.subscribe({ noAck: true }, (msg) => resolve(msg.bodyString()!))
  })

  expect(received).toBe("via subscribe")
  await session.stop()
})

test("AMQPQueue.publish() and get() confirmed round-trip", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const q = await session.queue("test-ws-spub-" + Math.random(), { durable: false, autoDelete: true })

  await q.publish("confirmed")
  const msg = await q.get({ noAck: true })
  expect(msg?.bodyString()).toBe("confirmed")

  await session.stop()
})

test("session.exchange() and AMQPExchange.publish() route messages", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const xName = "test-ws-x-" + Math.random()
  const q = await session.queue("test-ws-xq-" + Math.random(), { durable: false, autoDelete: true })

  const x = await session.fanoutExchange(xName, { durable: false, autoDelete: true })
  expect(x).toBeInstanceOf(AMQPExchange)
  await q.bind(xName)

  await x.publish("via exchange")
  await new Promise((resolve) => setTimeout(resolve, 100))

  const msg = await q.get({ noAck: true })
  expect(msg?.bodyString()).toBe("via exchange")

  await session.stop()
})

test("session.rpcClient() and session.rpcServer() round-trip", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const qName = "test-ws-rpc-" + Math.random()

  await session.rpcServer(qName, (msg) => {
    return `reply:${msg.bodyString()}`
  })

  const rpc = await session.rpcClient()
  const reply = await rpc.call(qName, "hello")
  expect(reply.bodyString()).toEqual("reply:hello")

  await session.stop()
})

test("session.rpcCall() one-shot round-trip", async () => {
  const session = await AMQPSession.connect(WS_URL)
  const qName = "test-ws-rpc-oneshot-" + Math.random()

  await session.rpcServer(qName, (msg) => {
    return `got:${msg.bodyString()}`
  })

  const reply = await session.rpcCall(qName, "ping")
  expect(reply.bodyString()).toEqual("got:ping")

  await session.stop()
})