//! The DoT listener (RFC 7858): the TCP/53 loop over a TLS stream.
//!
//! The slot pool, the accept loop and the shutdown protocol are
//! `listener.Core`'s (milestone-18 ruling 1), the same ones tcp_server uses.
//! What this file adds over TCP/53:
//!
//! - After the TCP accept, the certificate generation is pinned with
//!   `CertStore.acquire` and the mbedTLS handshake runs through
//!   `listener.handshakeStage` under the same race budget as every other
//!   per-connection operation, so a client that stalls mid-handshake cannot pin
//!   a connection slot.
//! - The framed-message loop reads and writes through
//!   `tls_server.ServerStream`, and closing the stream sends close_notify
//!   before the TCP close. A transport EOF without close_notify surfaces as a
//!   read failure, so a truncated connection is counted, never mistaken for a
//!   clean end.
//! - `ServerStream.accept` heap-allocates the mbedTLS ssl context, so unlike
//!   TCP/53 each connection costs one allocation. The pool itself is still
//!   fixed and pre-allocated.
//!
//! ALPN belongs to the `CertStore`'s `ServerContext` (set at store init);
//! this listener only acquires entries.

const std = @import("std");
const address = @import("../platform/address.zig");
const cert_store = @import("cert_store.zig");
const handler = @import("handler.zig");
const listener = @import("listener.zig");
const tls_server = @import("../platform/tls_server.zig");
const transport = @import("../upstream/transport.zig");
const upstream_owner = @import("../upstream/owner.zig");

/// Plaintext staging for `ServerStream`: the framing bytes and the decrypted
/// record tail pass through here, while whole messages go straight to
/// `Payload.query`/`Payload.reply`.
const stream_buffer_len = 1024;

pub const Options = struct {
    max_connections: u16 = 64,
    /// RFC 7766 §6.2.3 recommends a few seconds of idle tolerance, and the
    /// handshake runs under the same budget.
    idle_timeout: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake },
};

/// What DoT counts on top of `listener.CoreStats`: handshakes that failed or
/// outran the idle budget. `connections` in the core counts TCP connections
/// accepted, whether or not the handshake succeeded.
pub const Stats = struct {
    tls_handshake_failures: std.atomic.Value(u64) = .init(0),
};

/// The milestone-10 ruling 10 counters, the shape `metrics.counterGroup`
/// walks for the `nxdns_dot_server_*` families.
pub const StatsSnapshot = struct {
    connections: u64,
    tls_handshake_failures: u64,
    idle_timeouts: u64,
    connection_errors: u64,
};

pub const DotServer = struct {
    core: listener.Core(Config),
    handler: *handler.Handler,
    certs: *cert_store.CertStore,
    options: Options,
    stats: Stats,

    /// One slot is ~137 KiB — the same two message ceilings as TCP/53 plus the
    /// `ServerStream` bookkeeping — so the default 64 connections stay inside
    /// the PLAN §18 budget.
    pub const Payload = struct {
        query: [transport.max_message_len]u8,
        reply: [transport.max_message_len]u8,
        /// The handler's per-query working memory. It belongs to the slot so
        /// that answering a message allocates nothing, and a connection is
        /// answered serially, so one query uses it at a time.
        scratch: handler.Scratch,
        /// Pinned once its `accept` succeeds: mbedTLS holds a pointer to it,
        /// and the slot never moves.
        tls: tls_server.ServerStream,
    };

    const Config = struct {
        pub const Owner = DotServer;
        pub const ConnPayload = Payload;
        pub const serveConn = serveOne;
        pub const read_buffer_len = stream_buffer_len;
        pub const write_buffer_len = stream_buffer_len;
        pub const log = std.log.scoped(.dot_server);
        pub const name = "dot";
    };

    pub const Conn = listener.Core(Config).Conn;
    pub const ListenError = listener.Core(Config).ListenError;

    pub fn listen(
        gpa: std.mem.Allocator,
        io: std.Io,
        listen_address: std.Io.net.IpAddress,
        h: *handler.Handler,
        certs: *cert_store.CertStore,
        options: Options,
    ) ListenError!DotServer {
        return .{
            .core = try listener.Core(Config).listen(gpa, io, listen_address, options.max_connections),
            .handler = h,
            .certs = certs,
            .options = options,
            .stats = .{},
        };
    }

    /// The kernel-assigned address. A port of 0 in `listen` resolves here.
    pub fn boundAddress(self: *const DotServer) std.Io.net.IpAddress {
        return self.core.boundAddress();
    }

    /// Accept loop. Returns when the task is canceled or `deinit` stops it.
    pub fn serve(self: *DotServer, io: std.Io) void {
        self.core.serve(io);
    }

    pub fn deinit(self: *DotServer, io: std.Io) void {
        self.core.deinit(io);
        self.* = undefined;
    }

    pub fn snapshotStats(self: *const DotServer) StatsSnapshot {
        const core = &self.core.stats;
        return .{
            .connections = core.connections.load(.monotonic),
            .tls_handshake_failures = self.stats.tls_handshake_failures.load(.monotonic),
            .idle_timeouts = core.idle_timeouts.load(.monotonic),
            .connection_errors = core.connection_errors.load(.monotonic),
        };
    }

    /// One connection: pin, handshake, serve, close_notify, release — the
    /// ordering `listener.handshakeStage` documents. The core closes the TCP
    /// stream after this returns.
    fn serveOne(self: *DotServer, io: std.Io, index: usize) void {
        const conn = &self.core.conns[index];
        const stats = &self.core.stats;
        const gpa = self.core.gpa;
        const budget = self.options.idle_timeout;

        // Pins the certificate generation for the whole connection: a reload
        // that lands mid-stream retires the entry, and this reference keeps
        // the old context alive until the release below.
        const entry = self.certs.acquire(io);
        defer self.certs.release(io, entry);

        const stage: Handshake = .{ .conn = conn, .gpa = gpa, .ctx = &entry.ctx, .io = io };
        switch (listener.handshakeStage(io, budget, stage)) {
            .ok => {},
            .canceled => return,
            // A stalled handshake is refused like a broken one: it must not
            // pin a connection slot for longer than the idle budget.
            .timed_out, .failed => {
                listener.bump(&self.stats.tls_handshake_failures);
                return;
            },
        }
        // Sends close_notify and frees the ssl context; the core closes the
        // TCP stream afterwards.
        defer conn.payload.tls.close(gpa);

        const reader = conn.payload.tls.reader();
        const writer = conn.payload.tls.writer();

        while (true) {
            var prefix: [transport.prefix_len]u8 = undefined;
            var got: usize = 0;
            switch (listener.race(io, budget, listener.readPrefix, .{ reader, &prefix, &got })) {
                .ok => {},
                .timed_out => {
                    listener.bump(&stats.idle_timeouts);
                    return;
                },
                .canceled => return,
                // A transport EOF without close_notify lands here too: the
                // stream reads it as a truncation, never as a clean end.
                .failed => {
                    listener.bump(&stats.connection_errors);
                    return;
                },
            }

            // A client that sent close_notify between messages has finished
            // asking, which is the normal end of a connection, not a failure.
            if (got == 0) return;
            if (got != transport.prefix_len) {
                listener.bump(&stats.connection_errors);
                return;
            }

            // RFC 1035 §4.2.2 gives no meaning to a zero-length message, and
            // the prefix is a u16 so it can never exceed `max_message_len`.
            const len = transport.parsePrefix(prefix);
            if (len == 0) {
                listener.bump(&stats.connection_errors);
                return;
            }

            switch (listener.race(io, budget, listener.readBody, .{ reader, conn.payload.query[0..len] })) {
                .ok => {},
                .canceled => return,
                // A half-sent message is a broken peer, not an idle one.
                .timed_out, .failed => {
                    listener.bump(&stats.connection_errors);
                    return;
                },
            }

            const outcome = self.handler.handle(
                io,
                .tcp,
                address.NetAddress.fromIp(conn.peer),
                conn.payload.query[0..len],
                &conn.payload.reply,
                &conn.payload.scratch,
            );
            const bytes = switch (outcome) {
                // There is no framing for "no answer", so the connection ends.
                .drop => return,
                .reply => |b| b,
            };

            const out = transport.framePrefix(@intCast(bytes.len));
            switch (listener.race(io, budget, listener.writeReply, .{ writer, &out, bytes })) {
                .ok => {},
                .canceled => return,
                .timed_out, .failed => {
                    listener.bump(&stats.connection_errors);
                    return;
                },
            }
        }
    }

    /// The `listener.handshakeStage` stage: everything one mbedTLS handshake
    /// needs, plus the close that undoes it.
    const Handshake = struct {
        conn: *Conn,
        gpa: std.mem.Allocator,
        ctx: *tls_server.ServerContext,
        io: std.Io,

        pub fn accept(self: Handshake) anyerror!void {
            const conn = self.conn;
            try conn.payload.tls.accept(
                self.gpa,
                self.ctx,
                self.io,
                &conn.stream,
                &conn.read_buf,
                &conn.write_buf,
            );
        }

        pub fn close(self: Handshake) void {
            self.conn.payload.tls.close(self.gpa);
        }
    };
};

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------

const testing = std.testing;
const fixtures = @import("test_fixtures");
const header = @import("../dns/header.zig");
const model = @import("../config/model.zig");
const packet = @import("../dns/packet.zig");
const response = @import("../filter/response.zig");
const types = @import("../dns/types.zig");

test "snapshotStats reports the ruling-10 counters" {
    var server: DotServer = undefined;
    server.core.stats = .{};
    server.stats = .{};

    listener.bump(&server.core.stats.connections);
    listener.bump(&server.core.stats.connections);
    listener.bump(&server.stats.tls_handshake_failures);
    listener.bump(&server.core.stats.connection_errors);

    const snapshot = server.snapshotStats();
    try testing.expectEqual(@as(u64, 2), snapshot.connections);
    try testing.expectEqual(@as(u64, 1), snapshot.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), snapshot.idle_timeouts);
    try testing.expectEqual(@as(u64, 1), snapshot.connection_errors);
}

// -- loopback integration tests (gated on -Dintegration) --------------------
//
// Hermetic: one listener and one client on 127.0.0.1, an in-process fake
// upstream, and the fixture certificate written into a tmp directory for the
// `CertStore`. The whole client side of each test runs as one task raced
// against a budget so nothing can hang.

const blocking_defaults: model.Blocking = .{};
const blocking: response.Options = .{
    .mode = blocking_defaults.response,
    .ttl = blocking_defaults.ttl,
};
const forward_timeout: std.Io.Clock.Duration = .{
    .raw = model.readTimeout(.{}),
    .clock = .awake,
};

/// An upstream and nothing else optional: no filtering, no cache, no log. The
/// listener is what these tests exercise, so the handler is the same bare one
/// its own tests use.
fn bareHandler(up: *upstream_owner.Owner) handler.Handler {
    return .{
        .upstream = up,
        .policy = .{ .blocking = blocking, .forward_read_timeout = forward_timeout },
    };
}

const test_budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(5), .clock = .awake };

/// Short enough to keep the idle-timeout test quick, long enough that a
/// loopback handshake cannot lose to scheduling and time out on its own.
const short_idle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(300), .clock = .awake };

/// What the store advertises in production; std.crypto.tls.Client cannot send
/// ALPN in 0.16, so this also proves no-ALPN clients still connect.
const dot_alpn: [1:null]?[*:0]const u8 = .{"dot"};

/// A query for example.com A: id 0x1234, RD set, one question, no OPT.
const query_bytes =
    "\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00" ++
    "\x07example\x03com\x00\x00\x01\x00\x01";

/// The matching response: the question echoed plus one A record.
const response_bytes =
    "\x12\x34\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00" ++
    "\x07example\x03com\x00\x00\x01\x00\x01" ++
    "\xc0\x0c\x00\x01\x00\x01\x00\x00\x01\x2c\x00\x04\x5d\xb8\xd8\x22";

/// Answers from a fixture and rewrites the ID, which is all the server needs
/// from an upstream. The real clients are exercised by their own tests.
const FakeUpstream = struct {
    reply: []const u8,

    fn exchangeFn(
        ptr: *anyopaque,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError![]u8 {
        _ = io;
        selected.* = "fake://dot-server-upstream";
        const self: *FakeUpstream = @ptrCast(@alignCast(ptr));
        if (self.reply.len > response_buf.len) return error.ResponseTooLarge;
        @memcpy(response_buf[0..self.reply.len], self.reply);
        const bytes = response_buf[0..self.reply.len];
        packet.setId(bytes, (header.parse(query) catch unreachable).id);
        return bytes;
    }

    fn client(self: *FakeUpstream) transport.Client {
        return .{ .ptr = self, .exchangeFn = exchangeFn };
    }
};

/// Fixture PEMs written into a tmp directory and loaded into a `CertStore`,
/// addressed by cwd-relative paths the way `app.zig` hands config paths to
/// the store. Must not move after `init`: the store borrows the path slices.
const CertEnv = struct {
    tmp: testing.TmpDir,
    cert_path_buf: [128]u8,
    key_path_buf: [128]u8,
    store: cert_store.CertStore,

    fn init(env: *CertEnv, io: std.Io) !void {
        env.tmp = testing.tmpDir(.{});
        errdefer env.tmp.cleanup();

        try env.tmp.dir.writeFile(io, .{ .sub_path = "cert.pem", .data = fixtures.cert_pem });
        try env.tmp.dir.writeFile(io, .{ .sub_path = "key.pem", .data = fixtures.key_pem });
        const cert_path = try std.fmt.bufPrint(&env.cert_path_buf, ".zig-cache/tmp/{s}/cert.pem", .{env.tmp.sub_path});
        const key_path = try std.fmt.bufPrint(&env.key_path_buf, ".zig-cache/tmp/{s}/key.pem", .{env.tmp.sub_path});
        env.store = try cert_store.CertStore.init(testing.allocator, io, cert_path, key_path, &dot_alpn);
    }

    fn deinit(env: *CertEnv, io: std.Io) void {
        env.store.deinit(io);
        env.tmp.cleanup();
    }
};

const TestOutcome = union(enum) {
    work: anyerror!void,
    expiry: std.Io.Cancelable!void,
};

/// Runs the client side under a budget so a server that never answers fails
/// the test instead of hanging the run.
fn bounded(io: std.Io, comptime f: anytype, args: std.meta.ArgsTuple(@TypeOf(f))) !void {
    var outcomes: [2]TestOutcome = undefined;
    var select: std.Io.Select(TestOutcome) = .init(io, &outcomes);
    defer select.cancelDiscard();

    try select.concurrent(.work, f, args);
    try select.concurrent(.expiry, listener.expire, .{ io, test_budget });

    switch (try select.await()) {
        .work => |result| return result,
        .expiry => |result| {
            try result;
            return error.TestTimedOut;
        },
    }
}

/// The counters lag the client's last observable byte on some paths (a client
/// that drops its socket sees nothing after), so those tests wait for the
/// count instead of racing it.
fn waitForCounter(io: std.Io, counter: *std.atomic.Value(u64), want: u64) !void {
    const step: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(10), .clock = .awake };
    var attempts: usize = 0;
    while (counter.load(.monotonic) < want) : (attempts += 1) {
        if (attempts > 500) return error.TestTimedOut;
        step.sleep(io) catch return error.TestTimedOut;
    }
}

fn expectAnswersQuery(reply: []const u8) !void {
    const p = try packet.parse(reply);
    try testing.expectEqual(@as(u16, 0x1234), p.header.id);
    try testing.expectEqual(true, p.header.flags.qr);
    try testing.expectEqual(types.Rcode.no_error, p.header.flags.rcode);
    try testing.expectEqual(@as(u16, 1), p.header.ancount);

    const echoed = packet.firstQuestion(p) orelse return error.TestMissingQuestion;
    const asked = packet.firstQuestion(try packet.parse(query_bytes)).?;
    try testing.expectEqualSlices(u8, asked.name.wire(), echoed.name.wire());
    try testing.expectEqual(types.Type.a, echoed.qtype);
}

/// One TLS client over one TCP connection. Self-referential — the reader and
/// writer interfaces point into this struct — so it is pinned after `connect`.
const TestTls = struct {
    stream: std.Io.net.Stream,
    net_reader: std.Io.net.Stream.Reader,
    net_writer: std.Io.net.Stream.Writer,
    transport_read_buffer: [std.crypto.tls.Client.min_buffer_len]u8,
    transport_write_buffer: [std.crypto.tls.Client.min_buffer_len]u8,
    plaintext_read_buffer: [4096]u8,
    plaintext_write_buffer: [4096]u8,
    entropy: [std.crypto.tls.Client.Options.entropy_len]u8,
    client: std.crypto.tls.Client,

    fn connect(self: *TestTls, io: std.Io, address_: std.Io.net.IpAddress) !void {
        self.stream = try address_.connect(io, .{ .mode = .stream });
        errdefer self.stream.close(io);

        self.net_reader = self.stream.reader(io, &self.transport_read_buffer);
        self.net_writer = self.stream.writer(io, &self.transport_write_buffer);
        io.random(&self.entropy);

        self.client = try std.crypto.tls.Client.init(&self.net_reader.interface, &self.net_writer.interface, .{
            .host = .no_verification,
            .ca = .no_verification,
            .read_buffer = &self.plaintext_read_buffer,
            .write_buffer = &self.plaintext_write_buffer,
            .entropy = &self.entropy,
            .realtime_now = std.Io.Timestamp.now(io, .real),
        });
        try self.net_writer.interface.flush();
    }

    fn close(self: *TestTls, io: std.Io) void {
        self.stream.close(io);
    }

    fn sendQuery(self: *TestTls) !void {
        try self.client.writer.writeAll(&transport.framePrefix(@intCast(query_bytes.len)));
        try self.client.writer.writeAll(query_bytes);
        try self.client.writer.flush();
        try self.net_writer.interface.flush();
    }

    fn readReply(self: *TestTls) ![]u8 {
        const len = transport.parsePrefix((try self.client.reader.takeArray(transport.prefix_len)).*);
        return try self.client.reader.take(len);
    }
};

/// RFC 7766 §6.2.1.1 over TLS: two framed queries on one connection, answered
/// in order, then a clean close_notify exchange in both directions.
fn dotKeepAlive(io: std.Io, address_: std.Io.net.IpAddress) anyerror!void {
    var t: TestTls = undefined;
    try t.connect(io, address_);
    defer t.close(io);

    for (0..2) |_| {
        try t.sendQuery();
        try expectAnswersQuery(try t.readReply());
    }

    try t.client.end();
    try t.net_writer.interface.flush();

    // The server's close_notify must arrive as a clean end of stream.
    var tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, t.client.reader.readSliceAll(&tail));
}

/// One answered query, then the TCP connection drops with no close_notify.
fn dotDropWithoutCloseNotify(io: std.Io, address_: std.Io.net.IpAddress) anyerror!void {
    var t: TestTls = undefined;
    try t.connect(io, address_);
    defer t.close(io);

    try t.sendQuery();
    try expectAnswersQuery(try t.readReply());
}

/// Plain TCP bytes where a ClientHello belongs, then reads until the server
/// gives up (an alert followed by a close, or just the close).
fn dotGarbageHandshake(io: std.Io, address_: std.Io.net.IpAddress) anyerror!void {
    var stream = try address_.connect(io, .{ .mode = .stream });
    defer stream.close(io);

    var write_buf: [64]u8 = undefined;
    var writer = stream.writer(io, &write_buf);
    try writer.interface.writeAll("GET / HTTP/1.1\r\nHost: not-a-tls-client\r\n\r\n");
    try writer.interface.flush();

    var read_buf: [256]u8 = undefined;
    var reader = stream.reader(io, &read_buf);
    var sink: [256]u8 = undefined;
    while (true) {
        const n = reader.interface.readSliceShort(&sink) catch break;
        if (n == 0) break;
    }
}

/// Handshakes and then sends nothing, so the server must end the connection
/// on its own — with close_notify, which the client sees as a clean end.
fn dotIdleUntilServerCloses(io: std.Io, address_: std.Io.net.IpAddress) anyerror!void {
    var t: TestTls = undefined;
    try t.connect(io, address_);
    defer t.close(io);

    var tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, t.client.reader.readSliceAll(&tail));
}

test "dot: two framed queries share one TLS connection" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotKeepAlive, .{ io, server_address });

    // The client saw the server's close_notify, so the connection task has
    // already run to completion and the counters are final.
    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 0), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), stats.idle_timeouts);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);
    try testing.expectEqual(@as(u64, 2), h.stats.queries.load(.monotonic));

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}

test "dot: a transport EOF without close_notify is a connection error, not a crash" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotDropWithoutCloseNotify, .{ io, server_address });
    try waitForCounter(io, &server.core.stats.connection_errors, 1);

    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 0), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 1), stats.connection_errors);
    try testing.expectEqual(@as(u64, 1), h.stats.queries.load(.monotonic));

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}

test "dot: plain TCP bytes fail the handshake and are counted" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotGarbageHandshake, .{ io, server_address });
    try waitForCounter(io, &server.stats.tls_handshake_failures, 1);

    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 1), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);
    try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}

/// Ruling 6 across a live listener: a reload retires the old generation
/// without touching the connection that pinned it, and the next handshake is
/// served by the new one. std.crypto.tls.Client (0.16) frees the peer chain
/// inside `init` and keeps no field for it, so the identity swap is asserted
/// on the store's entry pointers instead of on the certificates the client saw.
fn dotReloadKeepsOldConnAndServesNew(io: std.Io, address_: std.Io.net.IpAddress, env: *CertEnv) anyerror!void {
    var first: TestTls = undefined;
    try first.connect(io, address_);
    defer first.close(io);

    try first.sendQuery();
    try expectAnswersQuery(try first.readReply());

    const old_entry = env.store.acquire(io);
    defer env.store.release(io, old_entry);

    try env.tmp.dir.writeFile(io, .{ .sub_path = "cert.pem", .data = fixtures.cert2_pem });
    try env.tmp.dir.writeFile(io, .{ .sub_path = "key.pem", .data = fixtures.key2_pem });
    try env.store.reload(io);

    const new_entry = env.store.acquire(io);
    defer env.store.release(io, new_entry);
    try testing.expect(old_entry != new_entry);

    // A handshake after the reload succeeds, served by the new generation.
    var second: TestTls = undefined;
    try second.connect(io, address_);
    defer second.close(io);
    try second.sendQuery();
    try expectAnswersQuery(try second.readReply());

    // The established connection still answers on the retired generation.
    try first.sendQuery();
    try expectAnswersQuery(try first.readReply());

    try second.client.end();
    try second.net_writer.interface.flush();
    var second_tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, second.client.reader.readSliceAll(&second_tail));

    try first.client.end();
    try first.net_writer.interface.flush();
    var first_tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, first.client.reader.readSliceAll(&first_tail));
}

test "dot: a reload serves new handshakes without breaking the old connection" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    // `CertStore.deinit` inside asserts refs == 0: both connections released
    // their generations before the clients saw close_notify.
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotReloadKeepsOldConnAndServesNew, .{ io, server_address, &env });

    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 2), stats.connections);
    try testing.expectEqual(@as(u64, 0), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), stats.idle_timeouts);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);
    try testing.expectEqual(@as(u64, 3), h.stats.queries.load(.monotonic));

    const store_stats = env.store.snapshotStats();
    try testing.expectEqual(@as(u64, 1), store_stats.reloads);
    try testing.expectEqual(@as(u64, 0), store_stats.reload_failures);

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}

/// S3.4 across a live listener: a cert PATH change — new files, not rewritten
/// ones — serves the new certificate on the next handshake while the
/// connection pinned to the old generation finishes on it.
fn dotPathChangeServesNewCert(io: std.Io, address_: std.Io.net.IpAddress, env: *CertEnv) anyerror!void {
    var first: TestTls = undefined;
    try first.connect(io, address_);
    defer first.close(io);

    try first.sendQuery();
    try expectAnswersQuery(try first.readReply());

    const old_entry = env.store.acquire(io);
    defer env.store.release(io, old_entry);

    try env.tmp.dir.writeFile(io, .{ .sub_path = "cert2.pem", .data = fixtures.cert2_pem });
    try env.tmp.dir.writeFile(io, .{ .sub_path = "key2.pem", .data = fixtures.key2_pem });
    var cert_buf: [128]u8 = undefined;
    var key_buf: [128]u8 = undefined;
    const next_cert = try std.fmt.bufPrint(&cert_buf, ".zig-cache/tmp/{s}/cert2.pem", .{env.tmp.sub_path});
    const next_key = try std.fmt.bufPrint(&key_buf, ".zig-cache/tmp/{s}/key2.pem", .{env.tmp.sub_path});

    const prepared = try env.store.preparePathChange(io, next_cert, next_key);
    env.store.publishPathChange(io, prepared);

    const new_entry = env.store.acquire(io);
    defer env.store.release(io, new_entry);
    try testing.expect(old_entry != new_entry);

    var second: TestTls = undefined;
    try second.connect(io, address_);
    defer second.close(io);
    try second.sendQuery();
    try expectAnswersQuery(try second.readReply());

    try first.sendQuery();
    try expectAnswersQuery(try first.readReply());

    try second.client.end();
    try second.net_writer.interface.flush();
    var second_tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, second.client.reader.readSliceAll(&second_tail));

    try first.client.end();
    try first.net_writer.interface.flush();
    var first_tail: [1]u8 = undefined;
    try testing.expectError(error.EndOfStream, first.client.reader.readSliceAll(&first_tail));
}

test "dot: a cert path change serves the new certificate on the next handshake" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotPathChangeServesNewCert, .{ io, server_address, &env });

    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 2), stats.connections);
    try testing.expectEqual(@as(u64, 0), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);

    const store_stats = env.store.snapshotStats();
    try testing.expectEqual(@as(u64, 1), store_stats.reloads);
    try testing.expectEqual(@as(u64, 0), store_stats.reload_failures);

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}

test "dot: an idle connection is closed with close_notify and counted" {
    const build_options = @import("build_options");
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var env: CertEnv = undefined;
    try env.init(io);
    defer env.deinit(io);

    var fake: FakeUpstream = .{ .reply = response_bytes };
    var h_owner: upstream_owner.Borrowed = .{};
    var h = bareHandler(h_owner.client(fake.client()));

    const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{
        .max_connections = 2,
        .idle_timeout = short_idle,
    });
    const server_address = server.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, DotServer.serve, .{ &server, io });

    try bounded(io, dotIdleUntilServerCloses, .{ io, server_address });

    const stats = server.snapshotStats();
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 1), stats.idle_timeouts);
    try testing.expectEqual(@as(u64, 0), stats.tls_handshake_failures);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);

    server.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
}