-- SPDX-FileCopyrightText: © 2022—2026 Vladimir Zorin <vladimir@deviant.guru>
-- SPDX-License-Identifier: LicenseRef-OWL-1.0-or-later
-- Licensed under OWL v1.0+. See LICENSE.

local testimony = require("testimony")
local lev = require("lev")
local message = require("dns.message")
local types = require("dns.types")

local testify = testimony.new("== lev ==")

-- =========================================================================
-- now()
-- =========================================================================

testify:that("now: returns a positive number", function()
	local t = lev.now()
	testimony.assert_true(type(t) == "number", "expected number, got " .. type(t))
	testimony.assert_true(t > 0, "expected positive, got " .. tostring(t))
end)

testify:that("now: is monotonically increasing", function()
	local t1 = lev.now()
	local t2 = lev.now()
	testimony.assert_true(t2 >= t1, "now() went backwards")
end)

-- =========================================================================
-- run() + spawn()
-- =========================================================================

testify:that("run: executes the main function", function()
	local executed = false
	lev.run(function()
		executed = true
	end)
	testimony.assert_true(executed, "main function was not executed")
end)

testify:that("run: returns main function results", function()
	local a, b = lev.run(function()
		return 42, "hello"
	end)
	testimony.assert_equal(42, a)
	testimony.assert_equal("hello", b)
end)

testify:that("run: returns nil + error on main function error", function()
	local ok, err = lev.run(function()
		error("boom")
	end)
	testimony.assert_nil(ok)
	testimony.assert_match("boom", err)
end)

testify:that("run: rejects nested calls", function()
	local ok, err = lev.run(function()
		lev.run(function() end)
	end)
	testimony.assert_nil(ok)
	testimony.assert_match("nested", err)
end)

testify:that("spawn: runs a concurrent coroutine", function()
	local order = {}
	lev.run(function()
		lev.spawn(function()
			order[#order + 1] = "spawned"
		end)
		order[#order + 1] = "main"
	end)
	-- Both should have run
	testimony.assert_equal(2, #order)
	-- Main runs first (it's the first spawned coroutine)
	testimony.assert_equal("main", order[1])
	testimony.assert_equal("spawned", order[2])
end)

testify:that("spawn: multiple spawns all execute", function()
	local count = 0
	lev.run(function()
		for i = 1, 10 do
			lev.spawn(function()
				count = count + 1
			end)
		end
	end)
	testimony.assert_equal(10, count)
end)

-- =========================================================================
-- sleep()
-- =========================================================================

testify:that("sleep: delays execution", function()
	local start, finish
	lev.run(function()
		start = lev.now()
		lev.sleep(0.05)
		finish = lev.now()
	end)
	local elapsed = finish - start
	testimony.assert_true(elapsed >= 0.04, "sleep too short: " .. tostring(elapsed))
	testimony.assert_true(elapsed < 0.5, "sleep too long: " .. tostring(elapsed))
end)

testify:that("sleep: concurrent sleeps run in parallel", function()
	local results = {}
	lev.run(function()
		local start = lev.now()
		local t1 = lev.spawn(function()
			lev.sleep(0.05)
			results[#results + 1] = { id = 1, elapsed = lev.now() - start }
		end)
		local t2 = lev.spawn(function()
			lev.sleep(0.05)
			results[#results + 1] = { id = 2, elapsed = lev.now() - start }
		end)
		-- Wait for both to finish by sleeping longer
		lev.sleep(0.15)
	end)
	testimony.assert_equal(2, #results)
	-- Both should have completed in roughly 50ms, not 100ms
	for _, r in ipairs(results) do
		testimony.assert_true(r.elapsed < 0.3, "task " .. r.id .. " took too long: " .. tostring(r.elapsed))
	end
end)

-- =========================================================================
-- await()
-- =========================================================================

testify:that("await: returns spawned task results", function()
	local result
	lev.run(function()
		local task = lev.spawn(function()
			return 42, "hello"
		end)
		result = { lev.await(task) }
	end)
	testimony.assert_equal(42, result[1])
	testimony.assert_equal("hello", result[2])
end)

testify:that("await: returns nil + error for failed task", function()
	local ok, err
	lev.run(function()
		local task = lev.spawn(function()
			error("task failed")
		end)
		ok, err = lev.await(task)
	end)
	testimony.assert_nil(ok)
	testimony.assert_match("task failed", err)
end)

testify:that("await: works when task already completed", function()
	local result
	lev.run(function()
		local task = lev.spawn(function()
			return "fast"
		end)
		-- Give it time to complete
		lev.sleep(0.01)
		result = lev.await(task)
	end)
	testimony.assert_equal("fast", result)
end)

testify:that("await: multiple awaiters on same task", function()
	local results = {}
	lev.run(function()
		local task = lev.spawn(function()
			lev.sleep(0.02)
			return "shared"
		end)
		lev.spawn(function()
			results[1] = lev.await(task)
		end)
		lev.spawn(function()
			results[2] = lev.await(task)
		end)
		lev.sleep(0.1)
	end)
	testimony.assert_equal("shared", results[1])
	testimony.assert_equal("shared", results[2])
end)

-- =========================================================================
-- cancel_token()
-- =========================================================================

testify:that("cancel_token: starts uncancelled", function()
	local token = lev.cancel_token()
	testimony.assert_false(token.cancelled, "token should start uncancelled")
end)

testify:that("cancel_token: cancel sets the flag", function()
	local token = lev.cancel_token()
	token:cancel()
	testimony.assert_true(token.cancelled, "token should be cancelled")
end)

testify:that("cancel_token: double cancel is safe", function()
	local token = lev.cancel_token()
	token:cancel()
	token:cancel()
	testimony.assert_true(token.cancelled)
end)

testify:that("cancel_token: cancellable sleep returns cancelled", function()
	local ok, err
	lev.run(function()
		local token = lev.cancel_token()
		lev.spawn(function()
			lev.sleep(0.01)
			token:cancel()
		end)
		ok, err = lev.sleep(10, token)
	end)
	testimony.assert_nil(ok)
	testimony.assert_equal("cancelled", err)
end)

testify:that("cancel_token: already-cancelled token short-circuits sleep", function()
	local ok, err
	lev.run(function()
		local token = lev.cancel_token()
		token:cancel()
		ok, err = lev.sleep(10, token)
	end)
	testimony.assert_nil(ok)
	testimony.assert_equal("cancelled", err)
end)

testify:that("cancel_token: mass cancel racing timer expiry does not corrupt the loop", function()
	-- Regression: token:cancel() used to enqueue a second resume (with extra
	-- stack args) for sleepers whose timers had already fired in the same loop
	-- iteration, corrupting the coroutine stack and panicking LuaJIT. The
	-- canceller registers its timer first, so with equal durations it wakes
	-- first in the same expiry sweep, while all sleepers sit in the ready
	-- queue — exactly the racy window.
	local results = {}
	local ok = lev.run(function()
		local token = lev.cancel_token()
		local canceller = lev.spawn(function()
			lev.sleep(0.05)
			token:cancel()
		end)
		local tasks = {}
		for i = 1, 46 do
			tasks[i] = lev.spawn(function()
				return lev.sleep(0.05, token)
			end)
		end
		lev.await(canceller)
		for i = 1, 46 do
			results[i] = { lev.await(tasks[i]) }
		end
		return true
	end)
	testimony.assert_true(ok, "loop did not complete cleanly")
	for i = 1, 46 do
		local r, err = results[i][1], results[i][2]
		testimony.assert_true(r == true or err == "cancelled", "task " .. i .. ": unexpected sleep result")
	end
end)

-- =========================================================================
-- race()
-- =========================================================================

testify:that("race: returns the first result", function()
	local result
	lev.run(function()
		result = lev.race({
			function(token)
				lev.sleep(0.01)
				return "fast"
			end,
			function(token)
				lev.sleep(1.0)
				return "slow"
			end,
		})
	end)
	testimony.assert_equal("fast", result)
end)

testify:that("race: cancels the token for losers", function()
	local token_was_cancelled = false
	lev.run(function()
		lev.race({
			function(token)
				return "winner"
			end,
			function(token)
				lev.sleep(0.05)
				token_was_cancelled = token.cancelled
				return "loser"
			end,
		})
		-- Let the loser finish to check its token state
		lev.sleep(0.1)
	end)
	testimony.assert_true(token_was_cancelled, "cancel token should be set for losers")
end)

testify:that("race: completes promptly with cancellable sleep", function()
	local elapsed
	lev.run(function()
		local start = lev.now()
		lev.race({
			function(token)
				lev.sleep(0.01, token)
				return "fast"
			end,
			function(token)
				lev.sleep(5.0, token)
				return "slow"
			end,
		})
		elapsed = lev.now() - start
	end)
	testimony.assert_true(elapsed < 0.5, "race took too long: " .. tostring(elapsed) .. "s (expected < 0.5s)")
end)

-- =========================================================================
-- all()
-- =========================================================================

testify:that("all: waits for all results", function()
	local results
	lev.run(function()
		results = lev.all({
			function(token)
				lev.sleep(0.01)
				return "a"
			end,
			function(token)
				lev.sleep(0.02)
				return "b"
			end,
		})
	end)
	testimony.assert_not_nil(results)
	testimony.assert_equal(2, #results)
	testimony.assert_equal("a", results[1][1])
	testimony.assert_equal("b", results[2][1])
end)

testify:that("all: fails fast on first error", function()
	local ok, err
	lev.run(function()
		ok, err = lev.all({
			function(token)
				error("fail early")
			end,
			function(token)
				lev.sleep(1.0)
				return "slow"
			end,
		})
	end)
	testimony.assert_nil(ok)
	testimony.assert_match("fail early", err)
end)

-- =========================================================================
-- UDP socket (basic)
-- =========================================================================

testify:that("udp: create and close without error", function()
	lev.run(function()
		local sock = lev.udp()
		testimony.assert_not_nil(sock)
		testimony.assert_true(sock:fd() >= 0, "expected valid fd")
		sock:close()
	end)
end)

testify:that("udp: bind to ephemeral port", function()
	lev.run(function()
		local sock = lev.udp()
		local ok, err = sock:bind("127.0.0.1", 0)
		testimony.assert_true(ok, "bind failed: " .. tostring(err))
		sock:close()
	end)
end)

testify:that("udp: setsockopt reuseaddr", function()
	lev.run(function()
		local sock = lev.udp()
		local ok, err = sock:setsockopt("reuseaddr", 1)
		testimony.assert_true(ok, "setsockopt failed: " .. tostring(err))
		sock:close()
	end)
end)

testify:that("udp: loopback sendto/recvfrom", function()
	lev.run(function()
		local server = lev.udp()
		server:setsockopt("reuseaddr", 1)
		server:bind("127.0.0.1", 19853)

		local client = lev.udp()

		-- Send from client to server
		local sent, err = client:sendto("hello", "127.0.0.1", 19853)
		testimony.assert_not_nil(sent, "sendto failed: " .. tostring(err))

		-- Receive on server
		local data, addr, port = server:recvfrom(1024, 1.0)
		testimony.assert_equal("hello", data)
		testimony.assert_equal("127.0.0.1", addr)
		testimony.assert_true(port > 0, "expected valid source port")

		server:close()
		client:close()
	end)
end)

testify:that("udp: recvfrom times out when no data", function()
	lev.run(function()
		local sock = lev.udp()
		sock:bind("127.0.0.1", 0)

		local data, err = sock:recvfrom(1024, 0.05)
		testimony.assert_nil(data)
		testimony.assert_equal("timeout", err)

		sock:close()
	end)
end)

-- =========================================================================
-- DNS query over lev.udp()
-- =========================================================================

testify:that("dns: single query via lev.udp resolves example.com", function()
	local resp_msg
	lev.run(function()
		local sock = lev.udp()

		local query_msg = {
			header = {
				id = 0xBEEF,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)

		local sent, err = sock:sendto(raw, "1.1.1.1", 53)
		testimony.assert_not_nil(sent, "sendto failed: " .. tostring(err))

		local data, addr, port = sock:recvfrom(4096, 5.0)
		testimony.assert_not_nil(data, "recvfrom failed: " .. tostring(addr))

		resp_msg = message.decode(data)
		sock:close()
	end)

	testimony.assert_not_nil(resp_msg, "failed to decode DNS response")
	testimony.assert_equal(1, resp_msg.header.qr)
	testimony.assert_equal(0xBEEF, resp_msg.header.id)
	testimony.assert_equal(0, resp_msg.header.rcode)
	testimony.assert_true(#resp_msg.answer > 0, "expected at least one answer")

	local found_a = false
	for _, rr in ipairs(resp_msg.answer) do
		if rr.type == types.TYPE.A then
			found_a = true
			testimony.assert_match("%d+%.%d+%.%d+%.%d+", rr.rdata.address)
		end
	end
	testimony.assert_true(found_a, "expected an A record in the answer")
end)

testify:that("dns: concurrent queries via lev.udp resolve in parallel", function()
	local results = {}
	lev.run(function()
		local domains = { "example.com.", "example.org.", "example.net." }
		local tasks = {}

		for i, domain in ipairs(domains) do
			tasks[i] = lev.spawn(function()
				local sock = lev.udp()
				local query_msg = {
					header = {
						id = 0x1000 + i,
						qr = 0,
						opcode = 0,
						rd = true,
						rcode = 0,
					},
					question = {
						{ name = domain, qtype = types.TYPE.A, qclass = types.CLASS.IN },
					},
				}
				local raw = message.encode(query_msg)

				local sent, err = sock:sendto(raw, "1.1.1.1", 53)
				if not sent then
					sock:close()
					return nil, "sendto: " .. tostring(err)
				end

				local data, addr, port = sock:recvfrom(4096, 5.0)
				sock:close()
				if not data then
					return nil, "recvfrom: " .. tostring(addr)
				end

				local resp = message.decode(data)
				if not resp then
					return nil, "decode failed"
				end
				return resp
			end)
		end

		for i, task in ipairs(tasks) do
			local resp, err = lev.await(task)
			results[i] = { resp = resp, err = err, domain = domains[i] }
		end
	end)

	testimony.assert_equal(3, #results)
	for _, r in ipairs(results) do
		testimony.assert_not_nil(r.resp, "query for " .. r.domain .. " failed: " .. tostring(r.err))
		testimony.assert_equal(1, r.resp.header.qr)
		testimony.assert_equal(0, r.resp.header.rcode)
		testimony.assert_true(#r.resp.answer > 0, "no answers for " .. r.domain)
	end
end)

-- =========================================================================
-- dns.transport.udp
-- =========================================================================

testify:that("lev_udp: query returns a valid DNS response", function()
	lev.run(function()
		local lev_udp = require("dns.transport.udp")
		local transport = lev_udp.new({ timeout = 5, udp_size = 4096 })

		-- Build a simple A query for example.com
		local query_msg = {
			header = {
				id = 0xCAFE,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)

		local data, err = transport:query(raw, "1.1.1.1", 53)
		testimony.assert_not_nil(data, "query failed: " .. tostring(err))

		local resp = message.decode(data)
		testimony.assert_not_nil(resp, "failed to decode response")
		testimony.assert_equal(1, resp.header.qr)
		testimony.assert_equal(0xCAFE, resp.header.id)
		testimony.assert_equal(0, resp.header.rcode)
		testimony.assert_true(#resp.answer > 0, "expected at least one answer")

		transport:close()
	end)
end)

testify:that("lev_udp: concurrent queries resolve in parallel", function()
	lev.run(function()
		local lev_udp = require("dns.transport.udp")
		local transport = lev_udp.new({ timeout = 5 })

		local domains = { "example.com.", "example.org.", "example.net." }
		local results = {}

		local tasks = {}
		for i, domain in ipairs(domains) do
			tasks[i] = lev.spawn(function()
				local query_msg = {
					header = {
						id = 0x2000 + i,
						qr = 0,
						opcode = 0,
						rd = true,
						rcode = 0,
					},
					question = {
						{ name = domain, qtype = types.TYPE.A, qclass = types.CLASS.IN },
					},
				}
				local raw = message.encode(query_msg)
				local data, err = transport:query(raw, "1.1.1.1", 53)
				if not data then
					return nil, err
				end
				return message.decode(data)
			end)
		end

		for i, task in ipairs(tasks) do
			local resp, err = lev.await(task)
			results[i] = { resp = resp, err = err, domain = domains[i] }
		end

		for _, r in ipairs(results) do
			testimony.assert_not_nil(r.resp, "query for " .. r.domain .. " failed: " .. tostring(r.err))
			testimony.assert_equal(0, r.resp.header.rcode)
		end

		transport:close()
	end)
end)

testify:that("lev_udp: timeout on unreachable server", function()
	lev.run(function()
		local lev_udp = require("dns.transport.udp")
		local transport = lev_udp.new({ timeout = 0.2 })

		local query_msg = {
			header = {
				id = 0xDEAD,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)

		-- 192.0.2.1 is TEST-NET, should be unreachable
		local data, err = transport:query(raw, "192.0.2.1", 53)
		testimony.assert_nil(data)
		testimony.assert_equal("timeout", err)

		transport:close()
	end)
end)

testify:that("lev_udp: works with dns.iter transport injection", function()
	lev.run(function()
		local lev_udp = require("dns.transport.udp")
		local iter = require("dns.iter")
		local transport = lev_udp.new({ timeout = 5, udp_size = 4096 })

		-- Use iter.query_server with injected lev transport
		local resp, err = iter.query_server("198.41.0.4", "example.com.", "A", {
			udp_transport = transport,
		})
		testimony.assert_not_nil(resp, "query_server failed: " .. tostring(err))
		testimony.assert_equal(1, resp.header.qr)

		transport:close()
	end)
end)

-- =========================================================================
-- TCP socket (basic)
-- =========================================================================

testify:that("tcp: create and close without error", function()
	lev.run(function()
		local sock = lev.tcp()
		testimony.assert_not_nil(sock)
		testimony.assert_true(sock:fd() >= 0, "expected valid fd")
		sock:close()
	end)
end)

testify:that("tcp: loopback connect + send/recv", function()
	lev.run(function()
		local listener = lev.listen("127.0.0.1", 0)
		testimony.assert_not_nil(listener, "listen failed")

		-- Get the port assigned by the OS
		-- We need to connect to the listener's fd to find the port
		-- Use getsockname trick: bind to port 0, kernel assigns an ephemeral port
		-- We'll use a fixed port instead for simplicity
		listener:close()

		local port = 19854
		listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener, "listen failed on port " .. port)

		-- Client connects in a separate coroutine
		local client_done = false
		local server_done = false

		lev.spawn(function()
			local client = lev.connect("127.0.0.1", port, { timeout = 5 })
			testimony.assert_not_nil(client, "connect failed")

			local sent, err = client:send("hello TCP")
			testimony.assert_not_nil(sent, "send failed: " .. tostring(err))

			local data, rerr = client:recv(1024, 5)
			testimony.assert_equal("echo:hello TCP", data)

			client:close()
			client_done = true
		end)

		-- Server accepts
		local peer, addr, port_num = listener:accept(5)
		testimony.assert_not_nil(peer, "accept failed: " .. tostring(addr))
		testimony.assert_equal("127.0.0.1", addr)

		local data, rerr = peer:recv(1024, 5)
		testimony.assert_equal("hello TCP", data)

		peer:send("echo:" .. data)
		-- Give client time to receive
		lev.sleep(0.05)
		peer:close()
		listener:close()
		server_done = true

		-- Wait for client
		lev.sleep(0.1)
		testimony.assert_true(client_done, "client did not complete")
	end)
end)

testify:that("tcp: recv times out when no data", function()
	lev.run(function()
		local port = 19855
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		lev.spawn(function()
			local client = lev.connect("127.0.0.1", port, { timeout = 5 })
			testimony.assert_not_nil(client)
			local data, err = client:recv(1024, 0.1)
			testimony.assert_nil(data)
			testimony.assert_equal("timeout", err)
			client:close()
		end)

		-- Accept but never send
		local peer = listener:accept(5)
		lev.sleep(0.3)
		if peer then
			peer:close()
		end
		listener:close()
	end)
end)

testify:that("tcp: recv_exactly accumulates partial reads", function()
	lev.run(function()
		local port = 19856
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		local result

		lev.spawn(function()
			local client = lev.connect("127.0.0.1", port, { timeout = 5 })
			testimony.assert_not_nil(client)
			-- Request exactly 10 bytes — server sends in two chunks
			result = client:recv_exactly(10, 5)
			client:close()
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		peer:send("ABCDE")
		lev.sleep(0.05)
		peer:send("FGHIJ")
		lev.sleep(0.1)
		peer:close()
		listener:close()

		lev.sleep(0.1)
		testimony.assert_equal("ABCDEFGHIJ", result)
	end)
end)

testify:that("tcp: recv_until finds pattern", function()
	lev.run(function()
		local port = 19857
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		local result

		lev.spawn(function()
			local client = lev.connect("127.0.0.1", port, { timeout = 5 })
			testimony.assert_not_nil(client)
			result = client:recv_until("\r\n\r\n", 5)
			client:close()
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		peer:send("Host: example.com\r\n")
		lev.sleep(0.02)
		peer:send("Content-Length: 0\r\n\r\n")
		lev.sleep(0.1)
		peer:close()
		listener:close()

		lev.sleep(0.1)
		testimony.assert_not_nil(result)
		testimony.assert_match("\r\n\r\n", result)
	end)
end)

testify:that("tcp: connect to unreachable host times out", function()
	lev.run(function()
		-- 192.0.2.1 is TEST-NET, should be unreachable
		local sock, err = lev.connect("192.0.2.1", 12345, { timeout = 0.2 })
		testimony.assert_nil(sock)
		testimony.assert_not_nil(err)
	end)
end)

testify:that("tcp: accept returns client address", function()
	lev.run(function()
		local port = 19858
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		lev.spawn(function()
			local client = lev.connect("127.0.0.1", port, { timeout = 5 })
			if client then
				lev.sleep(0.05)
				client:close()
			end
		end)

		local peer, addr, cport = listener:accept(5)
		testimony.assert_not_nil(peer)
		testimony.assert_equal("127.0.0.1", addr)
		testimony.assert_true(cport > 0, "expected valid client port")

		-- Also test getpeername on the accepted socket
		local paddr, pport = peer._raw:getpeername()
		testimony.assert_equal("127.0.0.1", paddr)
		testimony.assert_true(pport > 0)

		peer:close()
		listener:close()
	end)
end)

testify:that("tcp: concurrent connections", function()
	lev.run(function()
		local port = 19859
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		local N = 5
		local results = {}

		-- Server accepts N clients
		lev.spawn(function()
			for i = 1, N do
				local peer, addr = listener:accept(5)
				if peer then
					lev.spawn(function()
						local data = peer:recv(1024, 5)
						if data then
							peer:send("reply:" .. data)
						end
						lev.sleep(0.05)
						peer:close()
					end, { detached = true })
				end
			end
		end)

		-- Spawn N clients
		local tasks = {}
		for i = 1, N do
			tasks[i] = lev.spawn(function()
				local client = lev.connect("127.0.0.1", port, { timeout = 5 })
				if not client then
					return nil, "connect failed"
				end
				client:send("msg" .. i)
				local data = client:recv(1024, 5)
				client:close()
				return data
			end)
		end

		for i = 1, N do
			results[i] = lev.await(tasks[i])
		end

		listener:close()

		-- All should have received replies
		for i = 1, N do
			testimony.assert_not_nil(results[i], "client " .. i .. " got nil response")
			testimony.assert_match("^reply:", results[i])
		end
	end)
end)

testify:that("tcp: setsockopt nodelay", function()
	lev.run(function()
		local sock = lev.tcp()
		local ok, err = sock:setsockopt("nodelay", 1)
		testimony.assert_true(ok, "setsockopt nodelay failed: " .. tostring(err))
		sock:close()
	end)
end)

testify:that("tcp: connect supports cancel token", function()
	lev.run(function()
		local token = lev.cancel_token()
		lev.spawn(function()
			lev.sleep(0.05)
			token:cancel()
		end)
		-- 192.0.2.1 is TEST-NET, should be unreachable
		local sock, err = lev.connect("192.0.2.1", 12345, { timeout = 5, cancel = token })
		testimony.assert_nil(sock)
		testimony.assert_equal("cancelled", err)
	end)
end)

testify:that("tcp: accept supports cancel token via opts table", function()
	lev.run(function()
		local port = 19860
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)

		local token = lev.cancel_token()
		lev.spawn(function()
			lev.sleep(0.05)
			token:cancel()
		end)

		-- No one will connect, cancel should fire
		local peer, err = listener:accept({ timeout = 5, cancel = token })
		testimony.assert_nil(peer)
		testimony.assert_equal("cancelled", err)
		listener:close()
	end)
end)

-- =========================================================================
-- TLS tests
-- =========================================================================

testify:that("tls: connect to public TLS server (1.1.1.1:853)", function()
	lev.run(function()
		local sock, err = lev.connect("1.1.1.1", 853, {
			timeout = 5,
			tls = {
				mode = "client",
				verify = false,
			},
		})
		testimony.assert_not_nil(sock, "TLS connect failed: " .. tostring(err))
		sock:close()
	end)
end)

testify:that("tls: send/recv over TLS (DNS-over-TLS)", function()
	lev.run(function()
		local sock, err = lev.connect("1.1.1.1", 853, {
			timeout = 5,
			tls = {
				mode = "client",
				verify = false,
			},
		})
		testimony.assert_not_nil(sock, "TLS connect failed: " .. tostring(err))
		sock:setsockopt("nodelay", 1)

		-- Build a DNS query
		local query_msg = {
			header = {
				id = 0xABCD,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)
		local len = #raw

		-- DNS-over-TLS uses same 2-byte length prefix as TCP
		local bit = require("bit")
		local frame = string.char(bit.rshift(bit.band(len, 0xFF00), 8), bit.band(len, 0xFF)) .. raw
		local sent, serr = sock:send(frame)
		testimony.assert_not_nil(sent, "send failed: " .. tostring(serr))

		-- Receive 2-byte length prefix
		local len_bytes, rerr = sock:recv_exactly(2, 5)
		testimony.assert_not_nil(len_bytes, "recv length failed: " .. tostring(rerr))

		local resp_len = bit.lshift(string.byte(len_bytes, 1), 8) + string.byte(len_bytes, 2)
		testimony.assert_true(resp_len > 0, "empty response")

		-- Receive response body
		local resp_data, derr = sock:recv_exactly(resp_len, 5)
		testimony.assert_not_nil(resp_data, "recv body failed: " .. tostring(derr))

		local resp = message.decode(resp_data)
		testimony.assert_not_nil(resp, "failed to decode response")
		testimony.assert_equal(1, resp.header.qr)
		testimony.assert_equal(0xABCD, resp.header.id)
		testimony.assert_true(#resp.answer > 0, "expected at least one answer")

		sock:close()
	end)
end)

-- =========================================================================
-- DNS transport: tcp / tls
-- =========================================================================

testify:that("lev_tcp: DNS query over TCP", function()
	lev.run(function()
		local lev_tcp = require("dns.transport.tcp")
		local transport = lev_tcp.new({ timeout = 5 })

		local query_msg = {
			header = {
				id = 0x3001,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)

		local data, err = transport:query(raw, "1.1.1.1", 53)
		testimony.assert_not_nil(data, "TCP query failed: " .. tostring(err))

		local resp = message.decode(data)
		testimony.assert_not_nil(resp, "failed to decode response")
		testimony.assert_equal(1, resp.header.qr)
		testimony.assert_equal(0x3001, resp.header.id)
		testimony.assert_true(#resp.answer > 0, "expected at least one answer")

		transport:close()
	end)
end)

testify:that("lev_tcp: concurrent TCP queries", function()
	lev.run(function()
		local lev_tcp = require("dns.transport.tcp")
		local transport = lev_tcp.new({ timeout = 5 })

		local domains = { "example.com.", "example.org.", "example.net." }
		local results = {}

		local tasks = {}
		for i, domain in ipairs(domains) do
			tasks[i] = lev.spawn(function()
				local query_msg = {
					header = {
						id = 0x3000 + i,
						qr = 0,
						opcode = 0,
						rd = true,
						rcode = 0,
					},
					question = {
						{ name = domain, qtype = types.TYPE.A, qclass = types.CLASS.IN },
					},
				}
				local raw = message.encode(query_msg)
				local data, err = transport:query(raw, "1.1.1.1", 53)
				if not data then
					return nil, err
				end
				return message.decode(data)
			end)
		end

		for i, task in ipairs(tasks) do
			local resp, err = lev.await(task)
			results[i] = { resp = resp, err = err, domain = domains[i] }
		end

		for _, r in ipairs(results) do
			testimony.assert_not_nil(r.resp, "query for " .. r.domain .. " failed: " .. tostring(r.err))
			testimony.assert_equal(0, r.resp.header.rcode)
		end

		transport:close()
	end)
end)

testify:that("lev_tls: DNS-over-TLS query", function()
	lev.run(function()
		local lev_tls = require("dns.transport.tls")
		local transport = lev_tls.new({ timeout = 5, verify = false })

		local query_msg = {
			header = {
				id = 0x4001,
				qr = 0,
				opcode = 0,
				rd = true,
				rcode = 0,
			},
			question = {
				{ name = "example.com.", qtype = types.TYPE.A, qclass = types.CLASS.IN },
			},
		}
		local raw = message.encode(query_msg)

		local data, err = transport:query(raw, "1.1.1.1", 853)
		testimony.assert_not_nil(data, "TLS query failed: " .. tostring(err))

		local resp = message.decode(data)
		testimony.assert_not_nil(resp, "failed to decode response")
		testimony.assert_equal(1, resp.header.qr)
		testimony.assert_equal(0x4001, resp.header.id)
		testimony.assert_true(#resp.answer > 0, "expected at least one answer")

		transport:close()
	end)
end)

testify:that("lev_tcp: works with dns.iter transport injection", function()
	lev.run(function()
		local lev_tcp = require("dns.transport.tcp")
		local iter = require("dns.iter")
		local transport = lev_tcp.new({ timeout = 5 })

		local resp, err = iter.query_server("198.41.0.4", "example.com.", "A", {
			tcp_transport = transport,
		})
		testimony.assert_not_nil(resp, "query_server failed: " .. tostring(err))
		testimony.assert_equal(1, resp.header.qr)

		transport:close()
	end)
end)

-- =========================================================================
-- Deadlock detection
-- =========================================================================

testify:that("deadlock: detected when coroutines block with no I/O", function()
	-- Main function awaits a task that yields without registering any
	-- I/O or timers. Both coroutines are stuck with nothing to wake them.
	local ok, err = lev.run(function()
		local never = lev.spawn(function()
			coroutine.yield()
		end)
		lev.await(never)
	end)
	testimony.assert_nil(ok)
	testimony.assert_match("deadlock", err)
end)

-- =========================================================================
-- Review regression tests
-- =========================================================================

testify:that("spawn: errors outside lev.run()", function()
	local ok, err = pcall(lev.spawn, function() end)
	testimony.assert_true(not ok, "spawn outside run() should error")
	testimony.assert_match("outside", tostring(err))
end)

testify:that("await: task returning many values does not corrupt the stack", function()
	lev.run(function()
		local task = lev.spawn(function()
			local t = {}
			for i = 1, 200 do
				t[i] = i
			end
			return unpack(t, 1, 200)
		end)
		local r = { lev.await(task) }
		testimony.assert_equal(200, #r)
		testimony.assert_equal(200, r[200])
	end)
end)

testify:that("race: winner returning nil preserves arity", function()
	lev.run(function()
		local pack2 = function(...)
			return select("#", ...), ...
		end
		local cnt, a = pack2(lev.race({
			function()
				return nil
			end,
		}))
		testimony.assert_equal(1, cnt)
		testimony.assert_nil(a)

		local cnt2, b, c = pack2(lev.race({
			function()
				return nil, "x"
			end,
		}))
		testimony.assert_equal(2, cnt2)
		testimony.assert_nil(b)
		testimony.assert_equal("x", c)
	end)
end)

testify:that("race: erroring competitor is not logged as unawaited", function()
	local logged = {}
	lev.set_logger({
		log = function(_, msg)
			logged[#logged + 1] = tostring(msg.msg)
		end,
	})
	lev.run(function()
		local res, err = lev.race({
			function()
				error("boom-race")
			end,
			function(token)
				lev.sleep(1, token)
				return "slow"
			end,
		})
		testimony.assert_nil(res)
		testimony.assert_match("boom%-race", err)
	end)
	lev.set_logger(nil)
	for _, m in ipairs(logged) do
		testimony.assert_true(not m:match("unawaited"), "unexpected unawaited log: " .. m)
	end
end)

testify:that("tcp: close wakes a coroutine blocked on recv", function()
	local got_err
	lev.run(function()
		local port = 19861
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)
		lev.spawn(function()
			local peer = listener:accept(5)
			if peer then
				-- Hold the peer open; never send
				lev.sleep(0.5)
				peer:close()
			end
		end, { detached = true })
		local client = lev.connect("127.0.0.1", port, { timeout = 5 })
		testimony.assert_not_nil(client)
		local reader = lev.spawn(function()
			local data, err = client:recv(1024, 5)
			got_err = err
		end)
		lev.sleep(0.05) -- let the reader block on the fd
		client:close()
		lev.await(reader)
		listener:close()
	end)
	testimony.assert_equal("closed", got_err)
end)

testify:that("recv_until: timeout is an overall deadline, not per-chunk", function()
	lev.run(function()
		local port = 19862
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)
		lev.spawn(function()
			local peer = listener:accept(5)
			if peer then
				-- Trickle a byte every 50ms — each one resets a per-wait
				-- timeout, but must not extend an overall deadline
				for _ = 1, 20 do
					local ok = peer:send("a")
					if not ok then
						break
					end
					lev.sleep(0.05)
				end
				peer:close()
			end
		end, { detached = true })
		local client = lev.connect("127.0.0.1", port, { timeout = 5 })
		testimony.assert_not_nil(client)
		local start = lev.now()
		local data, err = client:recv_until("ZZZ", 0.3)
		local elapsed = lev.now() - start
		testimony.assert_nil(data)
		testimony.assert_equal("timeout", err)
		testimony.assert_true(elapsed < 0.8, "deadline overshoot: " .. tostring(elapsed))
		client:close()
		listener:close()
	end)
end)

testify:that("tcp: recv with negative maxlen falls back to default", function()
	lev.run(function()
		local port = 19863
		local listener = lev.listen("127.0.0.1", port)
		testimony.assert_not_nil(listener)
		lev.spawn(function()
			local peer = listener:accept(5)
			if peer then
				peer:send("negmax")
				lev.sleep(0.1)
				peer:close()
			end
		end, { detached = true })
		local client = lev.connect("127.0.0.1", port, { timeout = 5 })
		testimony.assert_not_nil(client)
		local data, err = client:recv(-1, 5)
		testimony.assert_equal("negmax", data)
		client:close()
		listener:close()
	end)
end)

testify:that("udp: recvfrom with negative maxlen falls back to default", function()
	lev.run(function()
		local a = lev.udp()
		local ok = a:bind("127.0.0.1", 19864)
		testimony.assert_true(ok, "bind failed")
		local b = lev.udp()
		b:sendto("udpneg", "127.0.0.1", 19864)
		local data = a:recvfrom(-1, 2)
		testimony.assert_equal("udpneg", data)
		a:close()
		b:close()
	end)
end)

-- =========================================================================
-- GC safety of the running coroutine
-- =========================================================================

testify:that("run: a lev coroutine survives GC while it drives a child coroutine", function()
	-- Regression: resume_coroutine released the lev coroutine's only registry
	-- reference before lua_resume(), and co_to_task references it only weakly.
	-- When that coroutine then drives a *nested* coroutine (exactly what a
	-- coroutine-based SAX/XML parser does — resume it repeatedly for events),
	-- the lev coroutine is suspended ("normal") and, being unrooted, a GC
	-- cycle triggered by the child's allocation collected it. Resuming back
	-- into the freed lev coroutine was a use-after-free that segfaulted.
	-- The lev coroutine must stay GC-rooted for the whole resume.
	local events = lev.run(function()
		-- Heavy-allocating producer that yields events, like lewis.xml.
		local producer = coroutine.create(function()
			for i = 1, 4000 do
				local junk = {}
				for j = 1, 200 do
					junk[j] = string.rep("x", 64) .. i .. ":" .. j
				end
				coroutine.yield(i, junk[1])
			end
		end)
		local n = 0
		while true do
			local ok, v = coroutine.resume(producer)
			if not ok or v == nil then
				break
			end
			n = n + 1
		end
		return n
	end)
	testimony.assert_equal(4000, events)
end)

testify:that("spawn: a child lev coroutine survives GC while driving its own nested coroutine", function()
	-- Same hazard for coroutines spawned onto the loop and resumed from the
	-- ready queue, not just the main one.
	local result = lev.run(function()
		local child = lev.spawn(function()
			local producer = coroutine.create(function()
				for i = 1, 2000 do
					local junk = {}
					for j = 1, 200 do
						junk[j] = string.rep("y", 64) .. i .. ":" .. j
					end
					coroutine.yield(i)
				end
			end)
			while true do
				local ok, v = coroutine.resume(producer)
				if not ok or v == nil then
					break
				end
			end
			return "child-survived"
		end)
		return (lev.await(child))
	end)
	testimony.assert_equal("child-survived", result)
end)

-- =========================================================================
-- Conclude
-- =========================================================================

testify:conclude()