-- 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.

--[==[
Asynchronous I/O runtime for Lilush, built on epoll. Provides
coroutine-based concurrency with timers, UDP/TCP sockets, cancellation
tokens, and structured task combinators.
]==]

local core = require("lev.core")

local g_loop = nil
local co_to_task = nil
local g_logger = nil
local g_task_id_seq = nil

-- ── constants ────────────────────────────────────────────────────────

local SIGTERM = 15
local SIGINT = 2
local SIGHUP = 1
local SIGKILL = 9
local SIGUSR1 = 10
local SIGUSR2 = 12

local DEFAULT_SEND_TIMEOUT = 30
local DEFAULT_TLS_TIMEOUT = 10
local DEFAULT_LISTEN_BACKLOG = 128
local DEFAULT_RECV_UNTIL_MAX = 65536
local RECV_UNTIL_CHUNK = 4096

-- ── logging ──────────────────────────────────────────────────────────

local log_error = function(msg)
	if g_logger then
		g_logger:log({ msg = msg, process = "lev" }, "error")
	else
		io.stderr:write("lev: " .. msg .. "\n")
	end
end

---! Inject a structured logger for LEV error reporting
---@ set_logger(`logger`{.tbl})
--[===[
Sets a `std.logger` instance to receive LEV error messages (cleanup
errors, detached task errors). The logger persists across `run()` calls.
Pass `nil` to revert to stderr.
]===]
local set_logger = function(logger)
	g_logger = logger
end

-- ── option parsing ───────────────────────────────────────────────────

local parse_opts = function(timeout_or_opts)
	if type(timeout_or_opts) == "table" then
		return timeout_or_opts.timeout, timeout_or_opts.cancel, timeout_or_opts.max_size
	end
	return timeout_or_opts, nil, nil
end

-- Pack varargs preserving trailing nils (# is unreliable with holes)
local pack_n = function(...)
	return { n = select("#", ...), ... }
end

-- ── task lifecycle ───────────────────────────────────────────────────

local on_complete = function(co, ok, ...)
	if not co_to_task then
		return
	end
	local task = co_to_task[co]
	if not task then
		return
	end
	-- Run registered cleanups in LIFO order
	if task.cleanups then
		for i = #task.cleanups, 1, -1 do
			local entry = task.cleanups[i]
			local ok2, err2
			if type(entry) == "function" then
				ok2, err2 = pcall(entry)
			elseif type(entry) == "table" and type(entry.close) == "function" then
				ok2, err2 = pcall(entry.close, entry)
			end
			if not ok2 then
				log_error("cleanup error: " .. tostring(err2))
			end
		end
		task.cleanups = nil
	end
	if ok then
		task.status = "done"
		task.results = { ... }
		task.nresults = select("#", ...)
	else
		task.status = "error"
		task.err = ...
		if task.detached or not task.awaited then
			log_error("unawaited task error: " .. tostring(task.err))
		end
	end
	for _, awaiter_co in ipairs(task.awaiters) do
		core.ready_enqueue(g_loop, awaiter_co, 0)
	end
	task.awaiters = {}
	co_to_task[co] = nil
	task.parent = nil
end

-- ── resource tracking ───────────────────────────────────────────────

---! Register a resource for auto-close when the current task completes
---@ own(`resource`{.tbl}) -> `resource`{.tbl}
--[===[
Registers `resource` for automatic cleanup when the current coroutine
finishes (success or error). The resource must have a `close()` method.
Returns the resource for chaining: `local sock = lev.own(lev.connect(...))`.
No-op outside `lev.run()` or from non-spawned coroutines.
]===]
local own = function(resource)
	if not co_to_task then
		return resource
	end
	local co = coroutine.running()
	if not co then
		return resource
	end
	local task = co_to_task[co]
	if not task then
		return resource
	end
	if not task.cleanups then
		task.cleanups = {}
	end
	task.cleanups[#task.cleanups + 1] = resource
	return resource
end

---! Register an arbitrary cleanup function for the current task
---@ defer(`fn`{.func})
--[===[
Registers `fn` to run when the current coroutine finishes (success or
error). Cleanups run in LIFO order, like Go's `defer`. Errors in cleanup
functions are logged via `log_error` and don't prevent other cleanups from running.
No-op outside `lev.run()` or from non-spawned coroutines.
]===]
local defer = function(fn)
	if not co_to_task then
		return
	end
	local co = coroutine.running()
	if not co then
		return
	end
	local task = co_to_task[co]
	if not task then
		return
	end
	if not task.cleanups then
		task.cleanups = {}
	end
	task.cleanups[#task.cleanups + 1] = fn
end

---! Remove a resource from auto-close tracking
---@ disown(`resource`{.tbl}) -> `resource`{.tbl}
--[===[
Removes `resource` from the current task's cleanup list (identity
comparison). Use for ownership transfer between coroutines. Returns
the resource for chaining. No-op if the resource is not tracked.
]===]
local disown = function(resource)
	if not co_to_task then
		return resource
	end
	local co = coroutine.running()
	if not co then
		return resource
	end
	local task = co_to_task[co]
	if not task or not task.cleanups then
		return resource
	end
	for i = #task.cleanups, 1, -1 do
		if task.cleanups[i] == resource then
			table.remove(task.cleanups, i)
			break
		end
	end
	return resource
end

-- ── spawn ────────────────────────────────────────────────────────────

---! Spawn a new concurrent task
---@ spawn(`fn`{.func}, `opts`{.tbl .opt}) -> `task`{.tbl}
--[===[
Spawns `fn` as a new coroutine that runs concurrently within the
current `lev.run()` event loop. Returns a task handle that can be
passed to `await()`.

Options:
- `detached` (boolean): If true, errors are logged when the task
  completes and the task is typically not awaited.
]===]
local spawn = function(fn, opts)
	if not g_loop then
		error("lev: spawn() outside lev.run()", 2)
	end
	opts = opts or {}
	local co = coroutine.create(fn)
	g_task_id_seq = g_task_id_seq + 1
	local parent_co = coroutine.running()
	local parent = parent_co and co_to_task[parent_co] or nil
	local task = {
		id = g_task_id_seq,
		name = opts.name,
		parent = parent,
		status = "pending",
		detached = opts.detached or false,
		results = nil,
		nresults = 0,
		err = nil,
		awaiters = {},
		awaited = false,
	}
	co_to_task[co] = task
	core.spawn(g_loop, co)
	return task
end

-- ── await ────────────────────────────────────────────────────────────

---! Wait for a spawned task to complete
---@ await(`task`{.tbl}) -> `result`{.tbl .or_nil}, `err`{.str .err}
--[===[
Yields the current coroutine until the given task finishes. Returns
the task's return values on success, or `nil, err` on error.
]===]
local await = function(task)
	task.awaited = true
	if task.status == "done" then
		return unpack(task.results, 1, task.nresults)
	elseif task.status == "error" then
		return nil, task.err
	end
	local co = coroutine.running()
	task.awaiters[#task.awaiters + 1] = co
	core.yield_to_loop()
	if task.status == "done" then
		return unpack(task.results, 1, task.nresults)
	else
		return nil, task.err
	end
end

-- ── sleep ────────────────────────────────────────────────────────────

---! Suspend the current coroutine for a duration
---@ sleep(`seconds`{.num}, `cancel`{.tbl .opt}) -> `true`{.bool .or_nil}, `err`{.str .err}
--[===[
Yields the current coroutine and resumes it after the given number
of seconds (fractional seconds supported). Returns `true` on
normal completion, or `nil, "cancelled"` if the cancel token has
fired by the time the coroutine wakes up.
]===]
local sleep = function(seconds, cancel)
	if not cancel then
		return core.timer_sleep(seconds)
	end
	if cancel.cancelled then
		return nil, "cancelled"
	end
	local timer_id = core.timer_register(seconds)
	local co = coroutine.running()
	cancel:register(co, -1, timer_id)
	local ok, err = core.yield_to_loop()
	cancel:deregister(co)
	-- The timer may expire in the same loop iteration as token:cancel();
	-- expiry wins that race in the core, so re-check the flag for
	-- deterministic semantics.
	if cancel.cancelled then
		return nil, "cancelled"
	end
	return ok, err
end

-- ── cancel token ─────────────────────────────────────────────────────

---! Create a cancellation token
---@ cancel_token() -> `token`{.tbl}
--[===[
Returns a cancel token that can be passed to `sleep()`, `connect()`,
`accept()`, and other blocking operations to enable cooperative
cancellation. Call `token:cancel()` to interrupt all registered waiters.
]===]
local token_cancel = function(self)
	if self.cancelled then
		return
	end
	self.cancelled = true
	for _, w in ipairs(self.waiters) do
		core.cancel_wait(w.co, w.fd, w.timer_id)
	end
	self.waiters = {}
end

local token_register = function(self, co, fd, timer_id)
	if self.cancelled then
		return false
	end
	self.waiters[#self.waiters + 1] = { co = co, fd = fd, timer_id = timer_id }
	return true
end

local token_deregister = function(self, co)
	for i, w in ipairs(self.waiters) do
		if w.co == co then
			table.remove(self.waiters, i)
			return
		end
	end
end

local cancel_token = function()
	return {
		cancelled = false,
		waiters = {},
		cancel = token_cancel,
		register = token_register,
		deregister = token_deregister,
	}
end

-- ── wait_with_cancel helper ──────────────────────────────────────────

local wait_with_cancel = function(wait_fn, fd, timeout, cancel)
	if cancel and cancel.cancelled then
		return nil, "cancelled"
	end
	local co = coroutine.running()
	if cancel then
		if not cancel:register(co, fd, 0) then
			return nil, "cancelled"
		end
	end
	local ok, err = wait_fn(fd, timeout)
	if cancel then
		cancel:deregister(co)
	end
	return ok, err
end

-- Public timeouts are overall deadlines for the whole operation; each
-- internal wait gets the remaining time. `deadline` is an absolute
-- core.now() timestamp, or nil for no limit.
local wait_until_deadline = function(wait_fn, fd, deadline, cancel)
	local timeout
	if deadline then
		timeout = deadline - core.now()
		if timeout <= 0 then
			return nil, "timeout"
		end
	end
	return wait_with_cancel(wait_fn, fd, timeout, cancel)
end

local deadline_for = function(timeout)
	return timeout and (core.now() + timeout) or nil
end

-- ── race ─────────────────────────────────────────────────────────────

---! Run functions concurrently, return first result
---@ race(`fns`{.tbl}) -> `results`{.tbl .or_nil}, `err`{.str .err}
--[===[
Spawns each function in the list. When the first one finishes (success
or error), cancels the rest and returns that result. Each function
receives a cancel token as its first argument.
]===]
local race
race = function(fns)
	if #fns == 0 then
		return nil, "race: no competitors"
	end
	local token = cancel_token()
	local winner_ok = nil
	local winner_results = nil
	local winner_nresults = 0
	local winner_err = nil
	local collector_co = coroutine.running()
	local done = false

	for _, fn in ipairs(fns) do
		local task = spawn(function()
			local results = pack_n(pcall(fn, token))
			if not done then
				done = true
				winner_ok = results[1]
				if winner_ok then
					winner_results = results
					winner_nresults = results.n - 1
				else
					winner_err = results[2]
				end
				token:cancel()
				core.ready_enqueue(g_loop, collector_co, 0)
			end
			if not results[1] then
				error(results[2], 0)
			end
			return unpack(results, 2, results.n)
		end)
		-- Errors are consumed by race's return value; without this every
		-- erroring competitor would also be logged as an unawaited error
		task.awaited = true
	end

	if not done then
		core.yield_to_loop()
	end

	if winner_ok == true then
		return unpack(winner_results, 2, winner_nresults + 1)
	elseif winner_ok == false then
		return nil, winner_err
	end
	return nil, "race: no winner"
end

-- ── all ──────────────────────────────────────────────────────────────

---! Run functions concurrently, wait for all to complete
---@ all(`fns`{.tbl}) -> `results`{.tbl .or_nil}, `err`{.str .err}
--[===[
Spawns each function and waits for all to complete. Returns a list
of result tables (each `{values...}`). If any task errors, cancels
the rest and returns `nil, err` immediately.
]===]
local all
all = function(fns)
	if #fns == 0 then
		return {}
	end
	local token = cancel_token()
	local tasks = {}
	local results = {}
	local total = #fns
	local completed = 0
	local first_err = nil
	local collector_co = coroutine.running()
	local done = false

	for i, fn in ipairs(fns) do
		tasks[i] = spawn(function()
			return fn(token)
		end)
	end

	-- Monitor each task via a watcher coroutine
	for i, task in ipairs(tasks) do
		spawn(function()
			await(task)
			if task.status == "error" and not done then
				done = true
				first_err = task.err
				token:cancel()
				core.ready_enqueue(g_loop, collector_co, 0)
				return
			end
			if task.status == "done" then
				results[i] = task.results
			end
			completed = completed + 1
			if completed >= total and not done then
				done = true
				core.ready_enqueue(g_loop, collector_co, 0)
			end
		end, { detached = true })
	end

	if not done then
		core.yield_to_loop()
	end

	if first_err then
		return nil, first_err
	end
	return results
end

-- ── on_signal ────────────────────────────────────────────────────────

---! Register a signal handler
---@ on_signal(`signum`{.num}, `handler`{.func}) -> `true`{.bool .or_nil}, `err`{.str .err}
--[===[
Registers a callback to be invoked when the specified signal is
delivered. Must be called within `lev.run()`. The handler receives
the signal number as its argument.
]===]
local on_signal = function(signum, handler)
	return core.signal_setup(signum, handler)
end

---! Force-stop the event loop
---@ stop()
--[===[
Sets the loop's `running` flag to false, causing `lev.run()` to exit
on the next iteration regardless of active coroutines. Intended for
use in signal handlers to implement graceful shutdown with a deadline.
Must be called within `lev.run()`.
]===]
local stop = function()
	if g_loop then
		core.loop_stop(g_loop)
	end
end

-- ── udp ──────────────────────────────────────────────────────────────

---! Create a non-blocking UDP socket
---@ udp() -> `socket`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a wrapped UDP socket that integrates with the event loop.
The `recvfrom` method yields the current coroutine until data
arrives, with optional timeout and cancellation support.
]===]
local udp = function()
	local raw, err = core.udp_new()
	if not raw then
		return nil, err
	end

	local self = {}

	self.bind = function(_, addr, port)
		return raw:bind(addr, port)
	end

	self.setsockopt = function(_, name, value)
		return raw:setsockopt(name, value)
	end

	self.sendto = function(_, data, addr, port)
		return raw:sendto(data, addr, port)
	end

	self.recvfrom = function(_, maxlen, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		local fd = raw:fd()

		-- Loop: epoll wakeups can be spurious for UDP (e.g. a datagram
		-- dropped after a checksum failure), so EAGAIN after a wait must
		-- retry, not leak to the caller
		while true do
			local data, addr, port = raw:recvfrom(maxlen)
			if data then
				return data, addr, port
			end
			if addr ~= "EAGAIN" then
				return nil, addr
			end
			local ok, wait_err = wait_until_deadline(core.wait_readable, fd, deadline, cancel)
			if not ok then
				return nil, wait_err
			end
		end
	end

	self.close = function(_)
		local fd = raw:fd()
		if fd >= 0 then
			core.fd_deregister(fd)
		end
		raw:close()
	end

	self.fd = function(_)
		return raw:fd()
	end

	return self
end

-- ── TLS identity ────────────────────────────────────────────────────

---! Parse a cert+key PEM pair into a reusable TLS identity
---@ tls_identity(`cert_path`{.str}, `key_path`{.str}, `hostname`{.str .opt}) -> `identity`{.udata .or_nil}, `err`{.str .err}
--[===[
Parses the PEM certificate chain and private key from files into
a Lua userdata that can be passed to `starttls()` via the `identity`
config field. The identity is parsed once and reused across connections,
avoiding per-connection file I/O and PEM parsing overhead.

The optional `hostname` is stored for SNI matching.
]===]
local tls_identity = function(cert_path, key_path, hostname)
	return core.tls_parse_identity(cert_path, key_path, hostname)
end

-- ── tcp ──────────────────────────────────────────────────────────────

---! Create a non-blocking TCP socket
---@ tcp() -> `socket`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a wrapped TCP socket that integrates with the event loop.
Provides `send`, `recv`, `recv_exactly`, `recv_until` methods
with async I/O via `wait_readable`/`wait_writable`. Supports TLS
upgrade via `starttls`.
]===]

local wrap_tcp

wrap_tcp = function(raw)
	local self = {}
	local recv_buf = ""

	self._raw = raw

	self.send = function(_, data, opts)
		opts = opts or {}
		local deadline = deadline_for(opts.timeout or DEFAULT_SEND_TIMEOUT)
		local cancel = opts.cancel
		local total = #data
		local sent = 0
		while sent < total do
			local n, err = raw:send(data:sub(sent + 1))
			if n then
				sent = sent + n
			elseif err == "EAGAIN" or err == "want_write" then
				local wok, werr = wait_until_deadline(core.wait_writable, raw:fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			else
				return nil, err
			end
		end
		-- TLS buffers whole records: the C send reports bytes consumed, not
		-- bytes on the wire. Drain the record buffer before claiming success,
		-- or the tail can be silently dropped at close. Immediate "done" for
		-- plain TCP.
		while true do
			local fstatus, ferr = raw:tls_flush()
			if fstatus == "done" then
				break
			end
			if not fstatus then
				return nil, ferr
			end
			local wok, werr = wait_until_deadline(core.wait_writable, raw:fd(), deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
		return sent
	end

	self.recv = function(_, maxlen, timeout_or_opts)
		if #recv_buf > 0 then
			local chunk
			if maxlen and maxlen < #recv_buf then
				chunk = recv_buf:sub(1, maxlen)
				recv_buf = recv_buf:sub(maxlen + 1)
			else
				chunk = recv_buf
				recv_buf = ""
			end
			return chunk
		end

		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)

		-- Loop until data: a single retry can leak "EAGAIN"/"want_read" to
		-- the caller (spurious wakeup, fragmented TLS record)
		while true do
			local data, err = raw:recv(maxlen)
			if data then
				return data
			end
			if err == "closed" then
				return nil, "closed"
			end
			if err ~= "EAGAIN" and err ~= "want_read" and err ~= "want_write" then
				return nil, err
			end
			local wait_fn = (err == "want_write") and core.wait_writable or core.wait_readable
			local wok, werr = wait_until_deadline(wait_fn, raw:fd(), deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
	end

	self.recv_exactly = function(_, n, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)

		while #recv_buf < n do
			if cancel and cancel.cancelled then
				return nil, "cancelled"
			end
			local data, err = raw:recv(n - #recv_buf)
			if data then
				recv_buf = recv_buf .. data
			else
				if err == "closed" then
					return nil, "closed"
				end
				if err ~= "EAGAIN" and err ~= "want_read" and err ~= "want_write" then
					return nil, err
				end
				local wait_fn = (err == "want_write") and core.wait_writable or core.wait_readable
				local wok, werr = wait_until_deadline(wait_fn, raw:fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			end
		end
		local result = recv_buf:sub(1, n)
		recv_buf = recv_buf:sub(n + 1)
		return result
	end

	self.recv_until = function(_, pattern, timeout_or_opts, max_size)
		local timeout, cancel, opts_max_size = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		max_size = max_size or opts_max_size or DEFAULT_RECV_UNTIL_MAX

		while true do
			local s, e = recv_buf:find(pattern)
			if s then
				local result = recv_buf:sub(1, e)
				recv_buf = recv_buf:sub(e + 1)
				return result
			end
			if #recv_buf > max_size then
				return nil, "max_size exceeded"
			end
			if cancel and cancel.cancelled then
				return nil, "cancelled"
			end
			local data, err = raw:recv(RECV_UNTIL_CHUNK)
			if data then
				recv_buf = recv_buf .. data
			else
				if err == "closed" then
					return nil, "closed"
				end
				if err ~= "EAGAIN" and err ~= "want_read" and err ~= "want_write" then
					return nil, err
				end
				local wait_fn = (err == "want_write") and core.wait_writable or core.wait_readable
				local wok, werr = wait_until_deadline(wait_fn, raw:fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			end
		end
	end

	self.starttls = function(_, config)
		config = config or {}
		local ok, err = raw:starttls_init(config)
		if not ok then
			return nil, err
		end
		local deadline = deadline_for(config.timeout or DEFAULT_TLS_TIMEOUT)
		local cancel = config.cancel
		while true do
			local status, serr = raw:starttls_step()
			if status == "done" then
				return true
			end
			if status == "want_read" then
				local wok, werr = wait_until_deadline(core.wait_readable, raw:fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			elseif status == "want_write" then
				local wok, werr = wait_until_deadline(core.wait_writable, raw:fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			else
				return nil, serr
			end
		end
	end

	self.setsockopt = function(_, name, value)
		return raw:setsockopt(name, value)
	end

	self.getpeername = function(_)
		return raw:getpeername()
	end

	self.shutdown = function(_, how)
		return raw:shutdown(how)
	end

	self.close = function(_)
		local fd = raw:fd()
		if fd >= 0 then
			core.fd_deregister(fd)
		end
		raw:close()
	end

	self.fd = function(_)
		return raw:fd()
	end

	return self
end

local tcp = function()
	local raw, err = core.tcp_new()
	if not raw then
		return nil, err
	end
	return wrap_tcp(raw)
end

---! Connect to a remote TCP server
---@ connect(`addr`{.str}, `port`{.num}, `opts`{.tbl .opt}) -> `socket`{.tbl .or_nil}, `err`{.str .err}
--[===[
Creates a TCP socket, connects to `addr:port`, and optionally
performs a TLS handshake. Returns a wrapped TCP socket on success.

Options:
- `timeout` (number): connection timeout in seconds
- `tls` (table): if present, perform TLS after connecting (passed to `starttls`)
- `cancel` (token): cancel token; propagated to TLS handshake if `tls.cancel` is not set
]===]
local connect = function(addr, port, opts)
	opts = opts or {}
	local sock, sock_err = tcp()
	if not sock then
		return nil, sock_err or "failed to create socket"
	end

	local ok, err = sock._raw:connect(addr, port)
	if not ok then
		if err ~= "EINPROGRESS" then
			sock:close()
			return nil, err
		end
		local wok, werr = wait_with_cancel(core.wait_writable, sock:fd(), opts.timeout, opts.cancel)
		if not wok then
			sock:close()
			return nil, werr
		end
		local cok, cerr = sock._raw:connect_finish()
		if not cok then
			sock:close()
			return nil, cerr
		end
	end

	if opts.tls then
		local tls_opts = {}
		for k, v in pairs(opts.tls) do
			tls_opts[k] = v
		end
		if opts.cancel and not tls_opts.cancel then
			tls_opts.cancel = opts.cancel
		end
		local tok, terr = sock:starttls(tls_opts)
		if not tok then
			sock:close()
			return nil, terr
		end
	end

	return sock
end

---! Create a bound listening TCP socket
---@ listen(`addr`{.str}, `port`{.num}, `opts`{.tbl .opt}) -> `listener`{.tbl .or_nil}, `err`{.str .err}
--[===[
Creates and binds a TCP listener socket. Returns a listener object
with an `accept` method that yields until a client connects.
`accept` takes a timeout (number) or an options table
`{ timeout = N, cancel = token }`.

Options:
- `reuseport` (boolean): enable SO_REUSEPORT
- `backlog` (number): listen backlog (default 128)
]===]
local listen = function(addr, port, opts)
	opts = opts or {}
	local sock, sock_err = tcp()
	if not sock then
		return nil, sock_err or "failed to create socket"
	end

	sock:setsockopt("reuseaddr", 1)
	if opts.reuseport then
		sock:setsockopt("reuseport", 1)
	end

	local ok, err = sock._raw:bind(addr, port)
	if not ok then
		sock:close()
		return nil, err
	end

	local lok, lerr = sock._raw:listen(opts.backlog or DEFAULT_LISTEN_BACKLOG)
	if not lok then
		sock:close()
		return nil, lerr
	end

	local listener = {}

	listener.accept = function(_, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		-- Loop: the pending connection can vanish between the epoll wakeup
		-- and accept() (client reset), yielding a spurious EAGAIN
		while true do
			local client_raw, caddr, cport = sock._raw:accept()
			if client_raw then
				return wrap_tcp(client_raw), caddr, cport
			end
			if caddr ~= "EAGAIN" then
				return nil, caddr
			end
			local wok, werr = wait_until_deadline(core.wait_readable, sock:fd(), deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
	end

	listener.close = function(_)
		sock:close()
	end

	listener.fd = function(_)
		return sock:fd()
	end

	return listener
end

-- ── AF_UNIX stream sockets ──────────────────────────────────────────
-- Reuse wrap_tcp: a connected unix stream behaves like a TCP stream for
-- send/recv/recv_until/shutdown/close/fd (and even starttls).

local unix_socket = function()
	local raw, err = core.unix_new()
	if not raw then
		return nil, err
	end
	return wrap_tcp(raw)
end

---! Connect to a Unix domain (AF_UNIX) stream socket
---@ connect_unix(`path`{.str}, `opts`{.tbl .opt}) -> `socket`{.tbl .or_nil}, `err`{.str .err}
--[===[
Creates an AF_UNIX stream socket and connects to `path`. A leading `@`
selects the Linux abstract namespace (no filesystem entry). Returns a
wrapped socket exposing the same methods as a TCP socket.

Options:
- `timeout` (number): connection timeout in seconds
- `tls` (table): if present, perform TLS after connecting (passed to `starttls`)
- `cancel` (token): cancel token; propagated to TLS handshake if `tls.cancel` is not set
]===]
local connect_unix = function(path, opts)
	opts = opts or {}
	local sock, sock_err = unix_socket()
	if not sock then
		return nil, sock_err or "failed to create socket"
	end

	local ok, err = sock._raw:connect_unix(path)
	if not ok then
		if err ~= "EINPROGRESS" then
			sock:close()
			return nil, err
		end
		local wok, werr = wait_with_cancel(core.wait_writable, sock:fd(), opts.timeout, opts.cancel)
		if not wok then
			sock:close()
			return nil, werr
		end
		local cok, cerr = sock._raw:connect_finish()
		if not cok then
			sock:close()
			return nil, cerr
		end
	end

	if opts.tls then
		local tls_opts = {}
		for k, v in pairs(opts.tls) do
			tls_opts[k] = v
		end
		if opts.cancel and not tls_opts.cancel then
			tls_opts.cancel = opts.cancel
		end
		local tok, terr = sock:starttls(tls_opts)
		if not tok then
			sock:close()
			return nil, terr
		end
	end

	return sock
end

---! Create a bound listening Unix domain (AF_UNIX) stream socket
---@ listen_unix(`path`{.str}, `opts`{.tbl .opt}) -> `listener`{.tbl .or_nil}, `err`{.str .err}
--[===[
Creates and binds an AF_UNIX stream listener at `path`. A leading `@`
selects the Linux abstract namespace (no filesystem entry, auto-cleaned).
Returns a listener object whose `accept` method yields until a client
connects, returning a wrapped socket (no peer address for AF_UNIX).
`accept` takes a timeout (number) or `{ timeout = N, cancel = token }`.

Options:
- `backlog` (number): listen backlog (default 128)
- `unlink` (boolean): for filesystem paths, remove a stale socket file before
  binding and on close (default true; ignored for abstract `@` paths)
]===]
local listen_unix = function(path, opts)
	opts = opts or {}
	local sock, sock_err = unix_socket()
	if not sock then
		return nil, sock_err or "failed to create socket"
	end

	local abstract = path:sub(1, 1) == "@"
	local unlink = (opts.unlink ~= false) and not abstract

	-- Clear a stale socket file so bind doesn't fail with EADDRINUSE.
	if unlink then
		os.remove(path)
	end

	local ok, err = sock._raw:bind_unix(path)
	if not ok then
		sock:close()
		return nil, err
	end

	local lok, lerr = sock._raw:listen(opts.backlog or DEFAULT_LISTEN_BACKLOG)
	if not lok then
		sock:close()
		if unlink then
			os.remove(path)
		end
		return nil, lerr
	end

	local listener = {}

	listener.accept = function(_, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		-- Loop: the pending connection can vanish between the epoll wakeup
		-- and accept() (client reset), yielding a spurious EAGAIN
		while true do
			local client_raw, cerr = sock._raw:accept_unix()
			if client_raw then
				return wrap_tcp(client_raw)
			end
			if cerr ~= "EAGAIN" then
				return nil, cerr
			end
			local wok, werr = wait_until_deadline(core.wait_readable, sock:fd(), deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
	end

	listener.close = function(_)
		sock:close()
		if unlink then
			os.remove(path)
		end
	end

	listener.fd = function(_)
		return sock:fd()
	end

	return listener
end

-- ── subprocess ──────────────────────────────────────────────────────

---! Execute an external command asynchronously
---@ exec(`path`{.str}, `args`{.tbl .opt}, `opts`{.tbl .opt}) -> `proc`{.tbl .or_nil}, `err`{.str .err}
--[===[
Spawns an external command as a child process with non-blocking pipe I/O
and pidfd-based exit notification. Returns a process handle with async
`send`, `recv`, `recv_stderr`, `read_all`, `wait`, and `kill` methods.

Options:
- `cwd` (string): working directory for child
- `env` (table): array of "KEY=VALUE" strings, replaces child environment
- `stderr_to_stdout` (boolean): merge stderr into stdout stream
]===]

local wrap_subprocess = function(raw)
	local self = {}
	self._raw = raw

	self.send = function(_, data, opts)
		opts = opts or {}
		local deadline = deadline_for(opts.timeout or DEFAULT_SEND_TIMEOUT)
		local cancel = opts.cancel
		local total = #data
		local sent = 0
		while sent < total do
			local n, err = raw:write(data:sub(sent + 1))
			if n then
				sent = sent + n
			elseif err == "EAGAIN" then
				local wok, werr = wait_until_deadline(core.wait_writable, raw:stdin_fd(), deadline, cancel)
				if not wok then
					return nil, werr
				end
			else
				return nil, err
			end
		end
		return sent
	end

	local recv_pipe = function(read_fn, fd_fn, maxlen, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)

		while true do
			local data, err = read_fn(raw, maxlen)
			if data then
				return data
			end
			if err ~= "EAGAIN" then
				return nil, err
			end

			local fd = fd_fn(raw)
			if fd < 0 then
				return nil, "closed"
			end
			local wok, werr = wait_until_deadline(core.wait_readable, fd, deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
	end

	self.recv = function(_, maxlen, timeout_or_opts)
		return recv_pipe(raw.read, raw.stdout_fd, maxlen, timeout_or_opts)
	end

	self.recv_stderr = function(_, maxlen, timeout_or_opts)
		return recv_pipe(raw.read_stderr, raw.stderr_fd, maxlen, timeout_or_opts)
	end

	self.read_all = function(_, timeout_or_opts, max_size)
		local timeout, cancel, opts_max_size = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		max_size = max_size or opts_max_size

		local chunks = {}
		local total = 0
		while true do
			local data, err = raw:read(65536)
			if data then
				total = total + #data
				if max_size and total > max_size then
					return nil, "read_all: max_size exceeded"
				end
				chunks[#chunks + 1] = data
			else
				if err == "closed" then
					break
				end
				if err ~= "EAGAIN" then
					return nil, err
				end
				local fd = raw:stdout_fd()
				if fd < 0 then
					break
				end
				local wok, werr = wait_until_deadline(core.wait_readable, fd, deadline, cancel)
				if not wok then
					return nil, werr
				end
			end
		end
		return table.concat(chunks)
	end

	self.wait = function(_, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)

		while true do
			local code, sig = raw:waitid_pidfd()
			if code then
				return code, sig
			end
			-- sig contains the error
			if sig ~= "EAGAIN" then
				return nil, sig
			end

			-- Wait for pidfd to become readable
			local fd = raw:pidfd()
			if fd < 0 then
				return nil, "closed"
			end
			local wok, werr = wait_until_deadline(core.wait_readable, fd, deadline, cancel)
			if not wok then
				return nil, werr
			end
		end
	end

	self.close_stdin = function(_)
		local fd = raw:stdin_fd()
		if fd >= 0 then
			core.fd_deregister(fd)
		end
		raw:close_stdin()
	end

	self.kill = function(_, sig)
		return raw:kill(sig)
	end

	self.pid = function(_)
		return raw:pid()
	end

	self.close = function(_)
		local stdout_fd = raw:stdout_fd()
		if stdout_fd >= 0 then
			core.fd_deregister(stdout_fd)
		end
		local stderr_fd = raw:stderr_fd()
		if stderr_fd >= 0 then
			core.fd_deregister(stderr_fd)
		end
		local stdin_fd = raw:stdin_fd()
		if stdin_fd >= 0 then
			core.fd_deregister(stdin_fd)
		end
		local pidfd = raw:pidfd()
		if pidfd >= 0 then
			core.fd_deregister(pidfd)
		end
		raw:close()
	end

	return self
end

local exec = function(path, args, opts)
	opts = opts or {}
	args = args or {}
	local cmd_name = path:match("^.*/([^/]+)$") or path
	local argv = { cmd_name }
	for _, a in ipairs(args) do
		-- The C core requires real strings (GC safety); numbers are fine to convert
		if type(a) == "number" then
			a = tostring(a)
		end
		argv[#argv + 1] = a
	end
	local raw, err = core.subprocess_spawn(path, argv, opts)
	if not raw then
		return nil, err
	end
	return wrap_subprocess(raw)
end

-- ── observability ────────────────────────────────────────────────────

---! Set or get the current task's name
---@ task_name(`name`{.str .opt}) -> `name`{.str .or_nil}
--[===[
With an argument, sets the name of the current task and returns it.
Without an argument, returns the current task's name (or nil).
No-op outside `lev.run()`.
]===]
local task_name = function(name)
	if not co_to_task then
		return nil
	end
	local co = coroutine.running()
	if not co then
		return nil
	end
	local task = co_to_task[co]
	if not task then
		return nil
	end
	if name ~= nil then
		task.name = name
	end
	return task.name
end

---! Get fast loop stats snapshot
---@ stats() -> `stats`{.tbl .or_nil}
--[===[
Returns a table with loop counters: `active_coros`, `registered_fds`,
`timer_count`, `ready_count` (from C core), and `task_count` (Lua-side
count of entries in the task registry). Returns nil outside `lev.run()`.
]===]
local stats = function()
	if not g_loop then
		return nil
	end
	local s = core.loop_stats(g_loop)
	local task_count = 0
	for _ in pairs(co_to_task) do
		task_count = task_count + 1
	end
	s.task_count = task_count
	return s
end

---! Get structured task list
---@ task_tree() -> `tasks`{.tbl .or_nil}
--[===[
Returns a sorted array of task info tables, each with fields:
`id`, `name`, `status`, `detached`, `parent_id`, `co_status`.
Sorted by `id` for stable output. Returns nil outside `lev.run()`.
]===]
local task_tree = function()
	if not co_to_task then
		return nil
	end
	local tasks = {}
	for co, task in pairs(co_to_task) do
		tasks[#tasks + 1] = {
			id = task.id,
			name = task.name,
			status = task.status,
			detached = task.detached,
			parent_id = task.parent and task.parent.id or nil,
			co_status = coroutine.status(co),
		}
	end
	table.sort(tasks, function(a, b)
		return a.id < b.id
	end)
	return tasks
end

---! Get detailed task dump with tracebacks
---@ task_dump() -> `tasks`{.tbl .or_nil}
--[===[
Like `task_tree()` but adds a `traceback` field for suspended coroutines,
showing the exact yield point. Has meaningful overhead -- use for debugging,
not monitoring. Returns nil outside `lev.run()`.
]===]
local task_dump = function()
	if not co_to_task then
		return nil
	end
	local tasks = {}
	for co, task in pairs(co_to_task) do
		local entry = {
			id = task.id,
			name = task.name,
			status = task.status,
			detached = task.detached,
			parent_id = task.parent and task.parent.id or nil,
			co_status = coroutine.status(co),
		}
		if coroutine.status(co) == "suspended" then
			entry.traceback = debug.traceback(co)
		end
		tasks[#tasks + 1] = entry
	end
	table.sort(tasks, function(a, b)
		return a.id < b.id
	end)
	return tasks
end

---! Enable SIGUSR1 task dump handler
---@ enable_dump_signal(`format`{.str .opt})
--[===[
Registers a SIGUSR1 handler that calls `task_dump()` and outputs the
result. Supports `"text"` (default) and `"json"` formats. Uses the
injected logger if set, otherwise stderr. Safe because signal handlers
in LEV run via `signalfd` dispatch (not async-signal context).
]===]
local enable_dump_signal = function(format)
	format = format or "text"
	on_signal(SIGUSR1, function()
		local dump = task_dump()
		if not dump then
			return
		end
		local output
		if format == "json" then
			local json = require("cjson.safe")
			output = json.encode(dump)
		else
			local lines = { "=== LEV task dump ===" }
			for _, t in ipairs(dump) do
				local name_part = t.name and (' "' .. t.name .. '"') or ""
				lines[#lines + 1] = string.format(
					"  task #%d%s  status=%s  co=%s  detached=%s  parent=%s",
					t.id,
					name_part,
					t.status,
					t.co_status,
					tostring(t.detached),
					t.parent_id and tostring(t.parent_id) or "none"
				)
				if t.traceback then
					for line in t.traceback:gmatch("[^\n]+") do
						lines[#lines + 1] = "    " .. line
					end
				end
			end
			lines[#lines + 1] = "=== end dump ==="
			output = table.concat(lines, "\n")
		end
		if g_logger then
			g_logger:log({ msg = output, process = "lev" }, "info")
		else
			io.stderr:write(output .. "\n")
		end
	end)
end

-- ── fd reader ─────────────────────────────────────────────────────────

---! Toggle O_NONBLOCK on an arbitrary fd
---@ set_nonblocking(`fd`{.num}, `enable`{.bool}) -> `was_nonblocking`{.bool .or_nil}, `err`{.str .err}
--[===[
Sets or clears `O_NONBLOCK` on `fd`. Returns the previous non-blocking
state so callers can restore it. Required before cooperatively reading
a shared fd (e.g. stdin) through the event loop.
]===]
local set_nonblocking = function(fd, enable)
	return core.set_nonblocking(fd, enable and true or false)
end

---! Wrap an arbitrary fd for cooperative reads on the event loop
---@ fd_reader(`fd`{.num}) -> `reader`{.tbl}
--[===[
Returns a reader for `fd` (e.g. stdin, fd 0) that integrates with the
event loop. `:read(maxlen, opts)` yields the current coroutine until at
least one byte is available, returning the bytes, or `nil,"closed"` on
EOF, or `nil,"cancelled"`/err. `:read_available(maxlen)` drains all
currently-readable bytes without yielding.

The fd must be in non-blocking mode (see `set_nonblocking`) and must be
epoll-able (a TTY or pipe, not a regular file). Only one coroutine may
wait on a given fd at a time.
]===]
local fd_reader = function(fd)
	local self = {}

	self.read = function(_, maxlen, timeout_or_opts)
		local timeout, cancel = parse_opts(timeout_or_opts)
		local deadline = deadline_for(timeout)
		while true do
			local data, err = core.read_fd(fd, maxlen)
			if data then
				return data
			end
			if err ~= "EAGAIN" then
				return nil, err
			end
			local ok, werr = wait_until_deadline(core.wait_readable, fd, deadline, cancel)
			if not ok then
				return nil, werr
			end
		end
	end

	self.read_available = function(_, maxlen)
		local parts = {}
		while true do
			local data, err = core.read_fd(fd, maxlen)
			if data then
				parts[#parts + 1] = data
			elseif err == "closed" then
				return table.concat(parts), "closed"
			else
				return table.concat(parts), "ok"
			end
		end
	end

	self.fd = function(_)
		return fd
	end

	self.close = function(_)
		if fd >= 0 then
			core.fd_deregister(fd)
		end
	end

	return self
end

-- ── run ──────────────────────────────────────────────────────────────

---! Run the event loop with a main function
---@ run(`fn`{.func}) -> `result`{.or_nil}, `err`{.str .err}
--[===[
Creates an event loop, spawns `fn` as the main coroutine, and runs
the loop until all tasks complete. Returns the main task's results
or `nil, err` if the main task errored.

Nested `lev.run()` calls are not allowed.
]===]
local run = function(fn)
	if g_loop then
		error("lev: nested lev.run() is not allowed")
	end
	g_loop = core.loop_new()
	co_to_task = setmetatable({}, { __mode = "k" })
	g_task_id_seq = 0

	local main_task = spawn(fn)
	main_task.name = main_task.name or "main"
	local loop_ok, loop_err = core.loop_run(g_loop, on_complete)

	local loop = g_loop
	g_loop = nil
	co_to_task = nil
	g_task_id_seq = nil
	core.loop_destroy(loop)

	if main_task.status == "done" then
		return unpack(main_task.results, 1, main_task.nresults)
	elseif main_task.status == "error" then
		return nil, main_task.err
	elseif loop_err then
		return nil, loop_err
	end
end

return {
	run = run,
	spawn = spawn,
	await = await,
	sleep = sleep,
	cancel_token = cancel_token,
	race = race,
	all = all,
	on_signal = on_signal,
	stop = stop,
	own = own,
	defer = defer,
	disown = disown,
	set_logger = set_logger,
	task_name = task_name,
	stats = stats,
	task_tree = task_tree,
	task_dump = task_dump,
	enable_dump_signal = enable_dump_signal,
	tls_identity = tls_identity,
	udp = udp,
	tcp = tcp,
	fd_reader = fd_reader,
	set_nonblocking = set_nonblocking,
	connect = connect,
	listen = listen,
	connect_unix = connect_unix,
	listen_unix = listen_unix,
	exec = exec,
	now = core.now,
	SIGTERM = SIGTERM,
	SIGINT = SIGINT,
	SIGHUP = SIGHUP,
	SIGKILL = SIGKILL,
	SIGUSR1 = SIGUSR1,
	SIGUSR2 = SIGUSR2,
}