// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>
// SPDX-License-Identifier: MIT

//! A `Connection` with the byte-shuffling done for you.
//!
//! This is the thin layer between the sans-I/O core and a stream: it moves
//! bytes out of a `std.Io.Writer` and into a `std.Io.Reader`, asks the clock
//! what time it is, and services the keepalive timer. It adds no protocol
//! knowledge whatsoever -- everything MQTT is in `mqtt.Connection`, in the
//! module this one depends on -- which is why it is about two hundred lines
//! and why anything it cannot do can be done by driving `Connection` directly.
//!
//! The reader and writer are the caller's. That is the whole point: a plain TCP
//! stream, a TLS session, a WebSocket carrying binary frames, a pair of pipes,
//! or a buffer in a test all present the same two interfaces, and this layer
//! cannot tell them apart.
//!
//! ```zig
//! var read_buffer: [4096]u8 = undefined;
//! var write_buffer: [4096]u8 = undefined;
//! var stream_reader = stream.reader(io, &read_buffer);
//! var stream_writer = stream.writer(io, &write_buffer);
//!
//! // The session is the caller's, because it outlives this transport and any
//! // number of later ones -- see `Reconnecting`, which relies on exactly that.
//! var conn: mqtt.Connection = .init(gpa, .{});
//! defer conn.deinit();
//!
//! var client: Client = .initOver(&conn, io, transport, streams);
//!
//! _ = try client.connect(.{ .client_id = "sensor-7" });
//! _ = try client.subscribe(&.{.{ .filter = "commands/#" }}, .empty);
//!
//! while (try client.next()) |event| switch (event) {
//!     .message => |m| std.debug.print("{s}: {s}\n", .{ m.topic, m.payload }),
//!     else => {},
//! }
//! ```
//!
//! **A plain `Client` cannot keep itself alive.** A `std.Io.Reader` has no
//! deadline, so `next` blocks in it for as long as the transport does -- and a
//! PINGREQ that falls due while it is blocked is never sent, so a broker
//! eventually decides the client is gone and closes the connection. On an idle
//! subscription that is a disconnection every keep-alive-and-a-half, which is
//! not a subtle failure but is a silent one.
//!
//! `initOver` is the answer: build the client over a `Transport` and every wait
//! is bounded by whenever the connection next wants the clock looked at. The
//! same constructor picks up the transport's flush, for one that buffers below
//! the writer it hands out -- neither transport here does, but a client built
//! over one that did and flushed only its own writer would send nothing at all.
//!
//! `Reconnecting` always uses it. `init` is for a reader and writer that came
//! from somewhere else entirely; there, keepalive is the caller's problem and
//! `Connection.deadline` says when it falls due.

const std = @import("std");
const mqtt = @import("mqtt");
const sasl = @import("sasl");

const Transport = @import("Transport.zig");

const Connection = mqtt.Connection;
const protocol = mqtt.protocol;
const packet = protocol.packet;

const PacketId = protocol.types.PacketId;
const Properties = protocol.properties.Properties;
const ReasonCode = protocol.types.ReasonCode;

const Client = @This();

pub const Event = Connection.Event;
pub const Message = Connection.Message;
pub const Options = Connection.Options;
pub const ConnectOptions = Connection.ConnectOptions;

pub const Error = Connection.Error || sasl.Client.Error ||
    std.Io.Reader.Error || std.Io.Writer.Error || error{
    /// The broker closed the transport in the middle of a packet.
    UnexpectedEndOfStream,
    /// The broker refused the connection. The CONNACK's reason code is in
    /// `refusal`.
    ConnectionRefused,
    /// The broker asked for an extended authentication exchange and no
    /// mechanism was given to answer it with. Set `authentication` before
    /// calling `connect`, or drive `Connection.auth` and `nextEvent` yourself.
    AuthenticationRequired,
    /// The broker accepted the connection, and the mechanism says it has not
    /// finished proving what it set out to prove.
    ///
    /// For a mechanism with mutual authentication -- SCRAM, DIGEST-MD5 -- this
    /// is the case the mutual half exists to catch: something answered, and it
    /// did not know the secret it would have needed to answer properly. The
    /// connection is not usable and has been disconnected.
    ServerNotAuthenticated,
    /// The response a mechanism wanted to send is longer than
    /// `Authentication.buffer`.
    AuthenticationDataTooLong,
    /// `reauthenticate` on a connection that never authenticated by a
    /// mechanism in the first place. Section 4.12.1 has nothing to say about
    /// that connection.
    NotAuthenticating,
};

/// How to answer a broker that wants more than a user name and a password.
///
/// MQTT 5.0 section 4.12 carries a SASL exchange in its own packets: the method
/// name goes in the CONNECT, each challenge arrives as an AUTH, and the last
/// message from the broker rides on the CONNACK. What it does *not* do is say
/// which mechanism -- so this takes `sasl.Client`, the interface, and the
/// mechanism is whatever the caller built.
///
/// ```zig
/// var plain: sasl.Plain = .init("user", "hunter2");
/// var buffer: [512]u8 = undefined;
/// client.authentication = .{ .mechanism = plain.client(), .buffer = &buffer };
/// ```
///
/// SCRAM is the same three lines through zig-scram's `scram-sasl` adapter,
/// and nothing here knows the difference.
pub const Authentication = struct {
    /// The mechanism, which runs **once**. A `Client` is one connection, so
    /// that is the right lifetime; `Reconnecting.Authenticator` is how to get
    /// a fresh one per attempt.
    mechanism: sasl.Client,
    /// Where each outgoing message is built. It only has to hold one message
    /// of whatever mechanism this is -- a few hundred bytes for SCRAM.
    buffer: []u8,
};

/// The session, which this borrows rather than owns.
///
/// Borrowing is what makes `Reconnecting` possible: a `Client` is one
/// transport, a `Connection` is a session, and a session outlives any number of
/// transports. It also keeps the layering honest -- everything MQTT is over
/// there, and a caller can reach past this at any point without unpicking
/// anything.
conn: *Connection,
io: std.Io,
reader: *std.Io.Reader,
writer: *std.Io.Writer,

/// The parts of the transport this needs beyond a reader and a writer: how to
/// wait with a deadline, and how to push bytes the writer has handed down but
/// not sent. See `next` and `flush`.
hooks: Transport.Hooks = .none,

/// Why the broker refused, kept from the CONNACK so that `connect` can return
/// an error and still say what happened.
refusal: ?Connection.Disconnected = null,

/// The mechanism to authenticate with, when the broker wants an exchange.
///
/// Set before `connect`. Left null, an AUTH from the broker comes back as
/// `error.AuthenticationRequired` and the exchange is the caller's to drive.
authentication: ?Authentication = null,

pub fn init(
    conn: *Connection,
    io: std.Io,
    reader: *std.Io.Reader,
    writer: *std.Io.Writer,
) Client {
    return .{ .conn = conn, .io = io, .reader = reader, .writer = writer };
}

/// A client over a transport, which is the form to prefer.
///
/// `streams` is what `transport.open` returned. The transport is asked for its
/// hooks rather than being kept, because the two things this needs from it --
/// waiting with a deadline, and pushing what is buffered below the writer --
/// are the only parts a client has any business with.
///
/// Both matter. Without the first, a `std.Io.Reader` has no deadline, so an
/// idle connection is eventually closed by the broker for never having pinged.
/// Without the second, a TLS session encrypts every packet into a buffer and
/// sends none of it.
pub fn initOver(
    conn: *Connection,
    io: std.Io,
    transport: Transport,
    streams: Transport.Streams,
) Client {
    return .{
        .conn = conn,
        .io = io,
        .reader = streams.reader,
        .writer = streams.writer,
        .hooks = transport.hooks(),
    };
}

/// The clock, in milliseconds, which is the only thing this layer asks the
/// `Io` for.
///
/// `.boot` rather than `.awake`, because it counts time the machine spent
/// suspended and the broker's keepalive timer certainly did: a laptop that
/// slept for an hour has been disconnected for an hour, and a client using a
/// clock that stopped with it would wake up believing it had just pinged.
fn now(c: *const Client) i64 {
    return std.Io.Timestamp.now(c.io, .boot).toMilliseconds();
}

/// Write everything the connection has queued, and flush it.
///
/// Called by everything else here, so a caller only needs it when driving the
/// connection directly between calls to this layer.
pub fn flush(c: *Client) Error!void {
    while (true) {
        const out = c.conn.pending();
        if (out.len == 0) break;
        try c.writer.writeAll(out);
        c.conn.sent(out.len, c.now());
    }
    try c.writer.flush();
    // And then whatever is below it. For a socket that is nothing; for a TLS
    // session it is the difference between having encrypted the packet and
    // having sent it.
    if (c.hooks.flush) |push| push(c.hooks.context, c.io) catch return error.WriteFailed;
}

/// What `nextBefore` found: one of three outcomes, where `next` has only two.
pub const Waited = union(enum) {
    event: Event,
    /// The broker closed the transport cleanly between packets, which is the
    /// normal end of a conversation that finished with a DISCONNECT.
    end_of_stream,
    /// Nothing arrived before the deadline. The connection is untouched and
    /// asking again carries on where this left off.
    timeout,
};

/// The next event, reading from the transport for as long as it takes.
///
/// Returns null when the broker closes the transport cleanly between packets,
/// which is the normal end of a conversation that finished with a DISCONNECT.
/// A transport that ends in the middle of a packet is `UnexpectedEndOfStream`.
pub fn next(c: *Client) Error!?Event {
    return switch (try c.nextBefore(.none)) {
        .event => |event| event,
        .end_of_stream => null,
        // `.none` is the one timeout that cannot expire.
        .timeout => unreachable,
    };
}

/// The next event, giving up at `timeout` and answering `.timeout` if nothing
/// has arrived by then.
///
/// For a caller that has something else to do on the same task. MQTT is one
/// connection carrying both directions, and this type is not safe to use from
/// two tasks at once -- `publish` and this both drive the same `Connection`,
/// which owns the transmit buffer, the packet identifier counter and the
/// in-flight window. A subscriber that also publishes therefore has to
/// interleave the two on one task, and an unbounded wait gives it no moment in
/// which to do so.
///
/// Keepalive is unaffected: the wait is the earlier of this deadline and
/// whenever the connection next wants the clock looked at, and a PINGREQ that
/// falls due while waiting goes out before this returns.
///
/// A deadline needs a client built with `initOver`, because the bound is
/// applied to the transport's `readable` hook. A plain `init` over a bare
/// reader has nothing to bound -- a `std.Io.Reader` cannot be asked to wait
/// only so long -- and the read blocks as it always did.
pub fn nextBefore(c: *Client, timeout: std.Io.Timeout) Error!Waited {
    // Resolved to an absolute instant once, here, rather than being applied on
    // each turn of the loop below: a duration restarted after every PINGREQ
    // would be a deadline that never arrives.
    // Measured as a remaining duration and added to this clock's own reading,
    // rather than with `Clock.Timestamp.toClock`: that is how a deadline on
    // any clock lands in the milliseconds-since-boot this file works in.
    const wait_until: ?i64 = switch (timeout.toDeadline(c.io)) {
        .none => null,
        .deadline => |at| c.now() + at.durationFromNow(c.io).raw.toMilliseconds(),
        // `toDeadline` returns one of the two above and never a duration.
        .duration => unreachable,
    };

    while (true) {
        if (try c.conn.nextEvent()) |event| {
            // Decoding may have queued a reply: the PUBACK owed for an incoming
            // QoS 1 message, a PUBREL, or the publications a just-resumed
            // session has to send again. Those go out now rather than waiting
            // for the next read to block -- a caller draining a burst of
            // buffered events would otherwise sit on its acknowledgements,
            // and a broker waiting for them would stop sending.
            //
            // Flushing does not touch the receive buffer, so the event being
            // returned is still good.
            if (c.conn.wantsWrite()) try c.flush();
            return .{ .event = event };
        }

        // Anything the connection wants to say, including a PINGREQ that has
        // just fallen due, goes out before the wait.
        try c.conn.tick(c.now());
        try c.flush();

        // With a waiter, the wait is bounded by whenever the connection next
        // wants the clock looked at, so a PINGREQ falling due in the middle of
        // an idle hour is actually sent. Without one the read below blocks
        // until the transport says otherwise, and an idle connection is
        // eventually closed by the broker for not having pinged -- which is
        // why `initOver` exists and why `Reconnecting` always uses it.
        if (c.hooks.readable) |readable| {
            while (c.reader.bufferedLen() == 0) {
                // Tested here, where the alternative is to wait, rather than
                // earlier: a caller polling with a deadline already past still
                // gets everything that has arrived, and still pings. Reaching
                // it a second time means the deadline that just expired was
                // the caller's and not the connection's, whose expiry the tick
                // below has already answered.
                if (c.expired(wait_until)) return .timeout;
                readable(c.hooks.context, c.io, c.waitTimeout(wait_until)) catch |err| switch (err) {
                    error.Timeout => {
                        try c.conn.tick(c.now());
                        try c.flush();
                        continue;
                    },
                    else => return error.ReadFailed,
                };
                break;
            }
        } else if (c.reader.bufferedLen() == 0 and c.expired(wait_until)) {
            // Nothing in hand and nothing to bound the read below, which would
            // block past the deadline and make a liar of it.
            return .timeout;
        }

        const available = c.reader.peekGreedy(1) catch |err| switch (err) {
            error.EndOfStream => {
                // Clean only if nothing was part-way through.
                if (c.conn.hasPartialPacket()) return error.UnexpectedEndOfStream;
                return .end_of_stream;
            },
            error.ReadFailed => return error.ReadFailed,
        };
        try c.conn.receive(available, c.now());
        c.reader.toss(available.len);
    }
}

/// Whether a caller's deadline, if there is one, has arrived.
fn expired(c: *const Client, wait_until: ?i64) bool {
    const until = wait_until orelse return false;
    return c.now() >= until;
}

/// How long to wait: until the connection wants the clock looked at again, or
/// until the caller's deadline, whichever comes first.
fn waitTimeout(c: *const Client, wait_until: ?i64) std.Io.Timeout {
    const deadline_ms = if (c.conn.deadline()) |wants|
        if (wait_until) |until| @min(wants, until) else wants
    else
        wait_until orelse return .none;
    const remaining = deadline_ms - c.now();
    return .{
        .duration = .{
            .raw = .fromMilliseconds(@max(0, remaining)),
            // The same clock the deadline was computed on, so that a machine which
            // suspends for an hour wakes up knowing the hour passed.
            .clock = .boot,
        },
    };
}

/// The next event that can be produced from bytes already in hand, without
/// waiting for the transport.
///
/// For a caller with an event loop of its own: poll the transport, `feed` what
/// it gave you, then drain this until it returns null.
pub fn nextBuffered(c: *Client) Error!?Event {
    return c.conn.nextEvent();
}

/// Hand over bytes read from the transport by somebody else.
pub fn feed(c: *Client, bytes: []const u8) Error!void {
    try c.conn.receive(bytes, c.now());
}

/// Send a CONNECT and wait for the CONNACK, carrying on an authentication
/// exchange if the broker asks for one.
///
/// Nothing can arrive before a CONNACK except an AUTH, so unlike the other
/// round trips here this one cannot swallow an event the caller wanted.
///
/// With `authentication` set, the whole of section 4.12 happens inside this
/// call: the method and the mechanism's first message go out in the CONNECT,
/// every AUTH is answered, and the data on the CONNACK -- which is where a
/// mechanism's *last* message arrives, the one where the broker proves itself
/// -- is fed back to the mechanism before this returns. Without it an AUTH is
/// reported as `error.AuthenticationRequired` and the conversation is the
/// caller's to hold with `Connection.auth` and `nextEvent`.
pub fn connect(c: *Client, options: ConnectOptions) Error!packet.ConnAck {
    try c.conn.connect(try c.prepareConnect(options), c.now());
    try c.flush();

    while (true) {
        const event = (try c.next()) orelse return error.UnexpectedEndOfStream;
        switch (event) {
            .connected => |ack| {
                c.finishAuthentication(ack) catch |err| {
                    // Nothing about this connection is worth keeping, and the
                    // broker is owed a DISCONNECT rather than a silence.
                    c.disconnect(.normal_disconnection, .empty) catch {};
                    return err;
                };
                return ack;
            },
            .disconnected => |d| {
                c.refusal = d;
                return error.ConnectionRefused;
            },
            .authenticate => |challenge| try c.answerChallenge(challenge),
            // Nothing else can precede a CONNACK, and `Connection` has already
            // refused anything that tried.
            else => unreachable,
        }
    }
}

/// Fill in what section 4.12 requires the CONNECT to carry, when there is a
/// mechanism to authenticate with.
///
/// The method names the mechanism and the data, if it has anything to say
/// first, is its opening message. `Initial.none` is a mechanism that will not
/// speak until spoken to, which here means sending the method and no data.
///
/// Public because `Reconnecting` sends its own CONNECT and needs the same
/// preparation; a caller driving `Connection` directly can use it too.
pub fn prepareConnect(c: *Client, options: ConnectOptions) Error!ConnectOptions {
    const a = if (c.authentication) |*auth_| auth_ else return options;
    if (options.properties.authentication_method != null) return error.AuthenticationMethodChanged;

    var opts = options;
    opts.properties.authentication_method = a.mechanism.name();
    var w: std.Io.Writer = .fixed(a.buffer);
    switch (a.mechanism.initial(&w) catch |err| return mapSasl(err)) {
        .written => opts.properties.authentication_data = w.buffered(),
        .none => {},
    }
    return opts;
}

/// Answer one AUTH by running the mechanism over the data it carried.
///
/// `error.AuthenticationRequired` when there is no mechanism, which is a broker
/// asking for a conversation nobody here can hold.
pub fn answerChallenge(c: *Client, challenge: packet.Auth) Error!void {
    const a = if (c.authentication) |*auth_| auth_ else return error.AuthenticationRequired;
    var w: std.Io.Writer = .fixed(a.buffer);
    // An AUTH with no data is a challenge of zero bytes, which several
    // mechanisms expect at least once and none can tell from a missing one.
    const data = challenge.properties.authentication_data orelse "";
    a.mechanism.respond(data, &w) catch |err| return mapSasl(err);
    try c.auth(.continue_authentication, .{ .authentication_data = w.buffered() });
}

/// Finish an exchange at the CONNACK, which is where MQTT puts the last
/// message of a mechanism that has one.
///
/// Most carriers have nowhere to put it -- SMTP says so outright -- and have to
/// deliver it as one more challenge answered with nothing. MQTT does not, which
/// is the one way this profile is easier than the others.
///
/// The `satisfied` check is what the mutual half of a mechanism exists for.
/// Something accepted the connection; that says it is willing to talk to us and
/// nothing at all about whether it is who it claims to be.
pub fn finishAuthentication(c: *Client, ack: packet.ConnAck) Error!void {
    const a = if (c.authentication) |*auth_| auth_ else return;
    if (ack.properties.authentication_data) |data| {
        var w: std.Io.Writer = .fixed(a.buffer);
        a.mechanism.respond(data, &w) catch |err| return mapSasl(err);
        // Whatever it wanted to say back is dropped: section 4.12 ends the
        // exchange at the CONNACK and there is no packet left to carry a
        // reply. Every mechanism that can finish here answers an empty
        // response anyway.
    }
    if (!a.mechanism.satisfied()) return error.ServerNotAuthenticated;
}

/// `sasl.Client.Error` is mostly `Io.Writer.Error`, which here means the
/// mechanism ran out of the buffer it was given rather than that a socket
/// broke -- and a caller told `WriteFailed` would go looking at the transport.
fn mapSasl(err: sasl.Client.Error) Error {
    return switch (err) {
        error.WriteFailed => error.AuthenticationDataTooLong,
        else => |e| e,
    };
}

/// Queue an AUTH and send it, continuing an authentication exchange.
///
/// The method is filled in by `Connection`, which knows what the CONNECT
/// named. For a caller driving an exchange by hand; `connect` uses it too.
pub fn auth(c: *Client, reason_code: ReasonCode, properties: Properties) Error!void {
    try c.conn.auth(reason_code, properties);
    try c.flush();
}

/// Authenticate again on a connection that is already up: section 4.12.1.
///
/// The mechanism in `authentication` runs from the beginning, so it must be a
/// fresh one -- a SASL exchange is a state machine that runs once, and the one
/// that established this connection has finished. Replace it first:
///
/// ```zig
/// var again: sasl.Plain = .init("user", "hunter2");
/// client.authentication = .{ .mechanism = again.client(), .buffer = &buffer };
/// try client.reauthenticate();
/// ```
///
/// The session is untouched and messages keep flowing throughout, which is
/// what section 4.12.1 is for: a credential that expires need not cost a
/// reconnection. Events that arrive while this is waiting are handled as they
/// would be by `next`, which means a message can be **dropped** -- like the
/// other round trips here, this is for a client that is not also a subscriber.
pub fn reauthenticate(c: *Client) Error!void {
    try c.beginReauthentication();

    while (true) {
        const event = (try c.next()) orelse return error.UnexpectedEndOfStream;
        switch (event) {
            .authenticate => |challenge| switch (challenge.reason_code) {
                .success => return c.finishReauthentication(challenge),
                else => try c.answerChallenge(challenge),
            },
            // Section 4.12.1: a broker that will not re-authenticate closes
            // the connection rather than refusing in place.
            .disconnected => |d| {
                c.refusal = d;
                return error.ConnectionRefused;
            },
            else => {},
        }
    }
}

/// Send the AUTH that starts a re-authentication and return without waiting.
///
/// For a caller that drives its own event loop and would rather not have this
/// one swallow what arrives while the exchange runs. `answerChallenge` handles
/// each `continue_authentication` and `finishReauthentication` the `success`
/// that ends it; `Reconnecting` does exactly that.
pub fn beginReauthentication(c: *Client) Error!void {
    const a: *Authentication = if (c.authentication) |*auth_| auth_ else return error.NotAuthenticating;
    var w: std.Io.Writer = .fixed(a.buffer);
    const initial = a.mechanism.initial(&w) catch |err| return mapSasl(err);
    try c.auth(.reauthenticate, switch (initial) {
        .written => .{ .authentication_data = w.buffered() },
        .none => .empty,
    });
}

/// Finish a re-authentication at the AUTH the broker ended it with.
///
/// The broker is content, which is what its `success` says. Whether we are is
/// a separate question and the one that matters -- the same check `connect`
/// makes at the CONNACK, for the same reason.
pub fn finishReauthentication(c: *Client, challenge: packet.Auth) Error!void {
    const a: *Authentication = if (c.authentication) |*auth_| auth_ else return error.NotAuthenticating;
    if (challenge.properties.authentication_data) |data| {
        var w: std.Io.Writer = .fixed(a.buffer);
        a.mechanism.respond(data, &w) catch |err| return mapSasl(err);
    }
    if (!a.mechanism.satisfied()) return error.ServerNotAuthenticated;
}

/// Publish a message, returning the packet identifier a later `.published`
/// event will carry, or null for QoS 0.
///
/// Returns as soon as the message is on the transport. Waiting for the
/// acknowledgement is the caller's business, because a client that waits cannot
/// have more than one message in flight and the flow control window exists
/// precisely so that it can.
pub fn publish(c: *Client, message: Message) Error!?PacketId {
    const id = try c.conn.publish(message);
    try c.flush();
    return id;
}

/// Subscribe, returning the packet identifier the SUBACK will carry.
///
/// The SUBACK arrives as a `.subscribed` event from `next`, along with whatever
/// else the broker has to say -- which will include messages on the new
/// subscription, possibly before the SUBACK itself.
pub fn subscribe(
    c: *Client,
    filters: []const packet.TopicFilter,
    properties: Properties,
) Error!PacketId {
    const id = try c.conn.subscribe(filters, properties);
    try c.flush();
    return id;
}

/// Unsubscribe, returning the packet identifier the UNSUBACK will carry.
pub fn unsubscribe(
    c: *Client,
    filters: []const []const u8,
    properties: Properties,
) Error!PacketId {
    const id = try c.conn.unsubscribe(filters, properties);
    try c.flush();
    return id;
}

/// Send a DISCONNECT and flush it.
///
/// The transport is the caller's to close, and should be closed after this: a
/// DISCONNECT that is still in a buffer when the socket goes is a DISCONNECT
/// that never happened, and the broker will publish the will message.
pub fn disconnect(c: *Client, reason_code: ReasonCode, properties: Properties) Error!void {
    try c.conn.disconnect(reason_code, properties);
    try c.flush();
}

/// Send a PINGREQ now. `next` sends these when they fall due; this is for a
/// caller that wants to know the broker is there.
pub fn ping(c: *Client) Error!void {
    try c.conn.ping(c.now());
    try c.flush();
}

/// The transport has gone. Keeps the session so that a new `Client` over a new
/// transport can resume it -- see `Connection.transportClosed`.
pub fn transportClosed(c: *Client) void {
    c.conn.transportClosed();
}

const testing = std.testing;

/// A pair of buffers pretending to be a transport, so that the layer can be
/// tested without a socket. Which is, after all, the claim it makes.
const Loopback = struct {
    to_client: std.Io.Reader,
    from_client: std.Io.Writer.Allocating,

    fn init(broker_says: []const u8) Loopback {
        return .{
            .to_client = .fixed(broker_says),
            .from_client = .init(testing.allocator),
        };
    }

    fn deinit(l: *Loopback) void {
        l.from_client.deinit();
    }
};

/// The bytes a broker would send to accept a connection: a CONNACK with nothing
/// to say.
const connack_bytes = [_]u8{ 0x20, 0x03, 0x00, 0x00, 0x00 };

test "a client connects over a pair of buffers" {
    var loop: Loopback = .init(&connack_bytes);
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    const ack = try client.connect(.{ .client_id = "test", .keep_alive = 0 });
    try testing.expect(!ack.session_present);
    try testing.expectEqual(ReasonCode.success, ack.reason_code);

    // And the CONNECT really did reach the writer.
    const written = loop.from_client.written();
    const header = try packet.peekFixedHeader(written);
    const sent_connect = try packet.Packet.decode(
        header,
        written[header.encodedLen()..header.packetLen()],
        .v5_0,
    );
    try testing.expectEqualStrings("test", sent_connect.connect.client_id);
}

test "a refused connection is an error that still says why" {
    // CONNACK with reason code 0x87, Not authorized.
    var loop: Loopback = .init(&.{ 0x20, 0x03, 0x00, 0x87, 0x00 });
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    try testing.expectError(error.ConnectionRefused, client.connect(.{ .client_id = "test" }));
    try testing.expectEqual(ReasonCode.not_authorized, client.refusal.?.reason_code);
    try testing.expectEqual(Connection.Disconnected.Origin.connack, client.refusal.?.origin);
}

test "messages are read out of the transport one event at a time" {
    // A CONNACK, then two QoS 0 publishes, then a DISCONNECT.
    const broker = connack_bytes ++ [_]u8{
        0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 'o', 'n', 'e', // publish "one"
        0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 't', 'w', 'o', // publish "two"
        0xe0, 0x00, // disconnect
    };
    var loop: Loopback = .init(&broker);
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    _ = try client.connect(.{ .client_id = "test", .keep_alive = 0 });
    try testing.expectEqualStrings("one", (try client.next()).?.message.payload);
    try testing.expectEqualStrings("two", (try client.next()).?.message.payload);
    try testing.expectEqual(
        ReasonCode.success,
        (try client.next()).?.disconnected.reason_code,
    );
    // The stream ends cleanly between packets, which is not an error.
    try testing.expectEqual(@as(?Event, null), try client.next());
}

test "a transport that ends in the middle of a packet is an error" {
    // The CONNACK, then a PUBLISH whose header promises ten bytes and delivers
    // three.
    const broker = connack_bytes ++ [_]u8{ 0x30, 0x0a, 0x00, 0x03 };
    var loop: Loopback = .init(&broker);
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    _ = try client.connect(.{ .client_id = "test", .keep_alive = 0 });
    try testing.expectError(error.UnexpectedEndOfStream, client.next());
}

test "an acknowledgement is sent before the message it acknowledges is handed over" {
    // A QoS 1 message and a QoS 0 one in the same read. Draining both without
    // flushing would leave the first one's PUBACK in the buffer while the
    // caller is told the message arrived -- and a broker with a small Receive
    // Maximum would then stop sending.
    const broker = connack_bytes ++ [_]u8{
        0x32, 0x0b, 0x00, 0x03, 'a', '/', 'b', 0x00, 0x2a, 0x00, 'o', 'n', 'e',
        0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 't',  'w',  'o',
    };
    var loop: Loopback = .init(&broker);
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    _ = try client.connect(.{ .client_id = "test", .keep_alive = 0 });
    const before = loop.from_client.written().len;

    try testing.expectEqualStrings("one", (try client.next()).?.message.payload);
    const after = loop.from_client.written();
    try testing.expect(after.len > before);

    // And what went out is the PUBACK for packet identifier 42.
    const header = try packet.peekFixedHeader(after[before..]);
    const ack = try packet.Packet.decode(
        header,
        after[before..][header.encodedLen()..header.packetLen()],
        .v5_0,
    );
    try testing.expectEqual(@as(mqtt.protocol.types.PacketId, 42), ack.puback.packet_id);

    try testing.expectEqualStrings("two", (try client.next()).?.message.payload);
}

test "publishing and subscribing reach the transport immediately" {
    var loop: Loopback = .init(&connack_bytes);
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    _ = try client.connect(.{ .client_id = "test", .keep_alive = 0 });
    const before = loop.from_client.written().len;

    _ = try client.publish(.{ .topic = "a/b", .payload = "x" });
    const after_publish = loop.from_client.written().len;
    try testing.expect(after_publish > before);

    const sub_id = try client.subscribe(&.{.{ .filter = "c/#" }}, .empty);
    try testing.expect(loop.from_client.written().len > after_publish);
    try testing.expect(sub_id != 0);
}

/// A reader that hands over one chunk per call and then says the stream ended.
///
/// `Reader.fixed` will not do for the keepalive test: it holds every byte from
/// the start, so the client never has to wait for anything and the waiting path
/// is never taken. What is needed is a stream where the second packet has not
/// arrived yet, which is what every real one looks like.
const ChunkedReader = struct {
    chunks: []const []const u8,
    index: usize = 0,
    interface: std.Io.Reader,
    buffer: [512]u8 = undefined,

    fn init(chunks: []const []const u8) ChunkedReader {
        return .{
            .chunks = chunks,
            .interface = .{
                .vtable = &.{ .stream = streamImpl },
                .buffer = &.{},
                .seek = 0,
                .end = 0,
            },
        };
    }

    /// Point the interface at this struct's own buffer. Done after the struct
    /// has stopped moving, because the interface holds a slice of it.
    fn attach(r: *ChunkedReader) *std.Io.Reader {
        r.interface.buffer = &r.buffer;
        return &r.interface;
    }

    fn streamImpl(
        io_r: *std.Io.Reader,
        w: *std.Io.Writer,
        limit: std.Io.Limit,
    ) std.Io.Reader.StreamError!usize {
        const r: *ChunkedReader = @alignCast(@fieldParentPtr("interface", io_r));
        if (r.index >= r.chunks.len) return error.EndOfStream;
        const chunk = r.chunks[r.index];
        r.index += 1;
        const dest = limit.slice(try w.writableSliceGreedy(1));
        const n = @min(dest.len, chunk.len);
        @memcpy(dest[0..n], chunk[0..n]);
        w.advance(n);
        return n;
    }
};

/// A waiter that reports nothing to read until it has done so once, sleeping
/// for as long as it was asked to wait.
///
/// That is what a real transport does when the broker is silent, and it is the
/// shape that used to break keepalive: the client blocked in the read, the
/// PINGREQ that fell due never went out, and the broker closed the connection
/// at a keep-alive and a half.
const IdleWaiter = struct {
    timeouts: usize = 0,

    fn readable(context: *anyopaque, io: std.Io, timeout: std.Io.Timeout) anyerror!void {
        const w: *IdleWaiter = @ptrCast(@alignCast(context));
        // `.none` means the connection has no timer running, which during the
        // handshake it does not: it is waiting for the CONNACK and nothing
        // else. A real waiter blocks; this one lets the reply through.
        if (timeout == .none) return;
        if (w.timeouts >= 1) return;
        w.timeouts += 1;
        timeout.sleep(io) catch {};
        return error.Timeout;
    }

    fn hooks(w: *IdleWaiter) Transport.Hooks {
        return .{ .context = w, .readable = readable };
    }
};

test "an idle connection pings rather than waiting to be disconnected" {
    // The broker says nothing after its CONNACK until it answers the ping, so
    // a client that never pinged would sit in the read until the transport gave
    // up -- which is how a real broker disconnected this client at a keep alive
    // and a half, and what the waiter is for. A keep alive of one second, so
    // the test costs about that.
    var reader: ChunkedReader = .init(&.{ &connack_bytes, &.{ 0xd0, 0x00 } });
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();

    var idle: IdleWaiter = .{};
    var client: Client = .{
        .conn = &conn,
        .io = testing.io,
        .reader = reader.attach(),
        .writer = &out.writer,
        .hooks = idle.hooks(),
    };

    _ = try client.connect(.{ .client_id = "idle", .keep_alive = 1 });
    const before = out.written().len;

    // The waiter reports nothing to read, which is where the ping has to
    // happen, and then lets the PINGRESP through.
    try testing.expectEqual(@as(Event, .pong), (try client.next()).?);
    try testing.expectEqual(@as(usize, 1), idle.timeouts);

    // And a PINGREQ really did go out in between.
    const after = out.written()[before..];
    const header = try packet.peekFixedHeader(after);
    try testing.expectEqual(mqtt.protocol.types.PacketType.pingreq, header.packet_type);
}

/// A waiter that never has anything to read: a broker that is connected and
/// silent, which is the case a deadline exists for.
const SilentWaiter = struct {
    timeouts: usize = 0,
    /// How many times to report nothing before letting the transport through.
    /// Null is a broker that never speaks again; zero is one with something to
    /// say already, so that a test can tell "did not wait" apart from "waited
    /// and heard nothing".
    limit: ?usize = null,

    fn readable(context: *anyopaque, io: std.Io, timeout: std.Io.Timeout) anyerror!void {
        const w: *SilentWaiter = @ptrCast(@alignCast(context));
        // The handshake waits on nothing in particular; let the CONNACK
        // through rather than deadlocking the connection.
        if (timeout == .none) return;
        if (w.limit) |limit| if (w.timeouts >= limit) return;
        w.timeouts += 1;
        timeout.sleep(io) catch {};
        return error.Timeout;
    }

    fn hooks(w: *SilentWaiter) Transport.Hooks {
        return .{ .context = w, .readable = readable };
    }
};

test "a deadline hands the task back rather than waiting for a silent broker" {
    var reader: ChunkedReader = .init(&.{&connack_bytes});
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();

    var silent: SilentWaiter = .{};
    var client: Client = .{
        .conn = &conn,
        .io = testing.io,
        .reader = reader.attach(),
        .writer = &out.writer,
        .hooks = silent.hooks(),
    };

    // A keep alive far longer than the deadline, so that what ends the wait
    // can only be the deadline.
    _ = try client.connect(.{ .client_id = "deadline", .keep_alive = 600 });

    const waited = try client.nextBefore(.{ .duration = .{
        .raw = .fromMilliseconds(30),
        .clock = .boot,
    } });
    try testing.expect(waited == .timeout);
    try testing.expect(silent.timeouts >= 1);

    // And the connection is untouched: asking again is not an error, and the
    // client still believes it is connected.
    const again = try client.nextBefore(.{ .duration = .{
        .raw = .fromMilliseconds(30),
        .clock = .boot,
    } });
    try testing.expect(again == .timeout);
}

test "a deadline already past is a poll rather than a wait" {
    var reader: ChunkedReader = .init(&.{ &connack_bytes, &.{ 0xd0, 0x00 } });
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();

    // Nothing to wait for: whatever the transport has is there to be read, so
    // a poll that consults the waiter at all would show up as a timeout here.
    var silent: SilentWaiter = .{ .limit = 0 };
    var client: Client = .{
        .conn = &conn,
        .io = testing.io,
        .reader = reader.attach(),
        .writer = &out.writer,
        .hooks = silent.hooks(),
    };

    _ = try client.connect(.{ .client_id = "poll", .keep_alive = 600 });
    try client.ping();

    const zero: std.Io.Timeout = .{ .duration = .{ .raw = .fromMilliseconds(0), .clock = .boot } };

    // Zero milliseconds is "do not wait", and the PINGRESP is one read away
    // rather than in hand -- so this answers without ever asking the waiter to
    // sleep, which is what makes such a poll cheap enough to do in a loop.
    try testing.expect((try client.nextBefore(zero)) == .timeout);
    try testing.expectEqual(@as(usize, 0), silent.timeouts);

    // The reply is still there, and a deadline with room in it finds it. The
    // poll consumed nothing and lost nothing.
    try testing.expectEqual(@as(Event, .pong), (try client.nextBefore(.{ .duration = .{
        .raw = .fromMilliseconds(500),
        .clock = .boot,
    } })).event);
}

test "a deadline does not suppress the keepalive it outlives" {
    // A keep alive of one second inside a deadline of about one and a half, so
    // the ping falls due while the caller is still waiting. The test costs
    // roughly that.
    var reader: ChunkedReader = .init(&.{&connack_bytes});
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();

    var silent: SilentWaiter = .{};
    var client: Client = .{
        .conn = &conn,
        .io = testing.io,
        .reader = reader.attach(),
        .writer = &out.writer,
        .hooks = silent.hooks(),
    };

    _ = try client.connect(.{ .client_id = "pinger", .keep_alive = 1 });
    const before = out.written().len;

    try testing.expect((try client.nextBefore(.{ .duration = .{
        .raw = .fromMilliseconds(1500),
        .clock = .boot,
    } })) == .timeout);

    // The wait was the earlier of the two deadlines each time round, so the
    // PINGREQ went out at a second and the caller was answered at a second and
    // a half.
    const after = out.written()[before..];
    const header = try packet.peekFixedHeader(after);
    try testing.expectEqual(mqtt.protocol.types.PacketType.pingreq, header.packet_type);
}

test "without the transport hooks the client still works, it just cannot ping" {
    // The plain constructor is not broken, only limited -- everything that does
    // not depend on the clock behaves identically.
    var loop: Loopback = .init(&(connack_bytes ++ [_]u8{ 0xd0, 0x00 }));
    defer loop.deinit();

    var conn: Connection = .init(testing.allocator, .{});
    defer conn.deinit();
    var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);

    _ = try client.connect(.{ .client_id = "plain", .keep_alive = 0 });
    try client.ping();
    try testing.expectEqual(@as(Event, .pong), (try client.next()).?);
    try testing.expect(client.hooks.readable == null);
    try testing.expect(client.hooks.flush == null);
}