-- 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 testify = testimony.new("== lev unix ==")

-- Distinct filesystem paths per test so a crashed run can't poison the next.
local SOCK = "/tmp/lilush_unix_test"

-- =========================================================================
-- connect_unix + listen_unix: basic echo
-- =========================================================================

testify:that("unix: loopback connect + send/recv", function()
	local path = SOCK .. "_echo.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener, "listen_unix failed")

		local client_done = false

		lev.spawn(function()
			local client, cerr = lev.connect_unix(path, { timeout = 5 })
			testimony.assert_not_nil(client, "connect_unix failed: " .. tostring(cerr))

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

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

			client:close()
			client_done = true
		end)

		local peer, aerr = listener:accept(5)
		testimony.assert_not_nil(peer, "accept failed: " .. tostring(aerr))

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

		peer:send("echo:" .. data)
		lev.sleep(0.05)
		peer:close()
		listener:close()

		lev.sleep(0.1)
		testimony.assert_true(client_done, "client did not complete")
	end)
	os.remove(path)
end)

-- =========================================================================
-- wrap_tcp reuse: recv_exactly / recv_until over a unix stream
-- =========================================================================

testify:that("unix: recv_exactly accumulates partial reads", function()
	local path = SOCK .. "_exactly.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)

		lev.spawn(function()
			local client = lev.connect_unix(path, { timeout = 5 })
			testimony.assert_not_nil(client)
			client:send("ab")
			lev.sleep(0.02)
			client:send("cdef")
			lev.sleep(0.05)
			client:close()
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		local data, err = peer:recv_exactly(6, 5)
		testimony.assert_equal("abcdef", data, "recv_exactly: " .. tostring(err))
		peer:close()
		listener:close()
	end)
	os.remove(path)
end)

testify:that("unix: recv_until matches a delimiter", function()
	local path = SOCK .. "_until.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)

		lev.spawn(function()
			local client = lev.connect_unix(path, { timeout = 5 })
			testimony.assert_not_nil(client)
			client:send("line one\nleftover")
			lev.sleep(0.05)
			client:close()
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		local data, err = peer:recv_until("\n", 5)
		testimony.assert_equal("line one\n", data, "recv_until: " .. tostring(err))
		peer:close()
		listener:close()
	end)
	os.remove(path)
end)

-- =========================================================================
-- Multiple concurrent connections to one listener
-- =========================================================================

testify:that("unix: multiple concurrent connections", function()
	local path = SOCK .. "_multi.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)

		local n = 5
		local got = {}

		-- Server: accept n clients, echo each
		lev.spawn(function()
			for _ = 1, n do
				local peer = listener:accept(5)
				if peer then
					lev.spawn(function()
						local d = peer:recv(1024, 5)
						peer:send("ok:" .. tostring(d))
						lev.sleep(0.02)
						peer:close()
					end, { detached = true })
				end
			end
		end, { detached = true })

		local tasks = {}
		for i = 1, n do
			tasks[i] = lev.spawn(function()
				local client = lev.connect_unix(path, { timeout = 5 })
				testimony.assert_not_nil(client, "client " .. i .. " connect failed")
				client:send("c" .. i)
				local reply = client:recv(1024, 5)
				client:close()
				return reply
			end)
		end

		for i = 1, n do
			got[i] = lev.await(tasks[i])
			testimony.assert_equal("ok:c" .. i, got[i])
		end

		listener:close()
	end)
	os.remove(path)
end)

-- =========================================================================
-- accept timeout / cancel
-- =========================================================================

testify:that("unix: accept times out when no client", function()
	local path = SOCK .. "_to.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)
		local peer, err = listener:accept(0.1)
		testimony.assert_nil(peer)
		testimony.assert_equal("timeout", err)
		listener:close()
	end)
	os.remove(path)
end)

testify:that("unix: accept is cancellable", function()
	local path = SOCK .. "_cancel.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)
		local token = lev.cancel_token()

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

		local peer, err = listener:accept({ timeout = 5, cancel = token })
		testimony.assert_nil(peer)
		testimony.assert_equal("cancelled", err)
		listener:close()
	end)
	os.remove(path)
end)

-- =========================================================================
-- Stale socket file is cleaned up before bind
-- =========================================================================

testify:that("unix: listen_unix removes a stale socket file", function()
	local path = SOCK .. "_stale.sock"
	os.remove(path)
	-- Bind with unlink=false so the socket file is left behind on close,
	-- becoming a stale entry that would otherwise make the next bind fail
	-- with EADDRINUSE.
	lev.run(function()
		local probe = lev.listen_unix(path, { unlink = false })
		testimony.assert_not_nil(probe)
		probe:close()
	end)
	-- A fresh listen with the default (unlink = true) must clear the stale
	-- file and succeed.
	lev.run(function()
		local l2, err = lev.listen_unix(path)
		testimony.assert_not_nil(l2, "listen over stale socket failed: " .. tostring(err))
		l2:close()
	end)
	os.remove(path)
end)

-- =========================================================================
-- Abstract namespace (no filesystem entry)
-- =========================================================================

testify:that("unix: abstract namespace bind/connect", function()
	local name = "@lilush_unix_test_abstract"
	lev.run(function()
		local listener = lev.listen_unix(name)
		testimony.assert_not_nil(listener, "abstract listen failed")

		local done = false
		lev.spawn(function()
			local client = lev.connect_unix(name, { timeout = 5 })
			testimony.assert_not_nil(client, "abstract connect failed")
			client:send("ping")
			local reply = client:recv(1024, 5)
			testimony.assert_equal("pong", reply)
			client:close()
			done = true
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		local d = peer:recv(1024, 5)
		testimony.assert_equal("ping", d)
		peer:send("pong")
		lev.sleep(0.05)
		peer:close()
		listener:close()
		lev.sleep(0.05)
		testimony.assert_true(done, "abstract client did not complete")
	end)
end)

-- =========================================================================
-- Error paths
-- =========================================================================

testify:that("unix: connect to nonexistent path errors", function()
	local path = SOCK .. "_nope.sock"
	os.remove(path)
	lev.run(function()
		local client, err = lev.connect_unix(path, { timeout = 1 })
		testimony.assert_nil(client)
		testimony.assert_not_nil(err, "expected an error string")
	end)
end)

testify:that("unix: path too long errors cleanly", function()
	local long = "/tmp/" .. string.rep("x", 200) .. ".sock"
	lev.run(function()
		local client, err = lev.connect_unix(long)
		testimony.assert_nil(client)
		testimony.assert_equal("path too long", err)

		local listener, lerr = lev.listen_unix(long)
		testimony.assert_nil(listener)
		testimony.assert_equal("path too long", lerr)
	end)
end)

-- =========================================================================
-- Double-close idempotency
-- =========================================================================

testify:that("unix: double close is safe", function()
	local path = SOCK .. "_dclose.sock"
	os.remove(path)
	lev.run(function()
		local listener = lev.listen_unix(path)
		testimony.assert_not_nil(listener)

		lev.spawn(function()
			local client = lev.connect_unix(path, { timeout = 5 })
			testimony.assert_not_nil(client)
			client:close()
			client:close() -- second close must not error
		end)

		local peer = listener:accept(5)
		testimony.assert_not_nil(peer)
		peer:close()
		peer:close() -- idempotent
		lev.sleep(0.05)
		listener:close()
		listener:close() -- idempotent
	end)
	os.remove(path)
end)

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

testify:conclude()