//! A bounded, caller-driven 9P2000 server connection.
//! The backend owns fids, authentication, permissions, and filesystem operations.
//! receive() borrows one input frame until release(). Pending operations may outlive
//! that frame only if the backend copies their strings/data. reply() copies output.
const std = @import("std");
const assert = std.debug.assert;
const wire = @import("wire.zig");
const Server = @This();

in: []u8,
out: []u8,
in_len: usize = 0,
frame: u32 = 0,
out_len: usize = 0,
out_off: usize = 0,
msize: u32 = 0,
dead: bool = false,
pending: [65]Pending = @splat(.{}),
versioning: bool = false,

const Pending = struct {
    request: ?wire.Type = null,
    tag: u16 = 0,
    count: u32 = 0,
    oldtag: u16 = wire.notag,
};

pub const Error = wire.Error || error{ Protocol, NoTags, UnknownTag, WrongReply, TooLarge };
pub const Options = struct { in: []u8, out: []u8 };
/// Local resource floor, not a minimum imposed by the 9P specification.
pub const msize_min: u32 = 24;

pub fn init(options: Options) Server {
    assert(options.in.len >= msize_min);
    assert(options.out.len >= msize_min);
    return .{ .in = options.in, .out = options.out };
}

pub fn push(s: *Server, bytes: []const u8) usize {
    if (s.dead) return 0;
    const n = @min(bytes.len, s.in.len - s.in_len);
    @memcpy(s.in[s.in_len..][0..n], bytes[0..n]);
    s.in_len += n;
    return n;
}

pub fn output(s: *const Server) []const u8 {
    return s.out[s.out_off..s.out_len];
}

pub fn wrote(s: *Server, n: usize) void {
    assert(n <= s.output().len);
    s.out_off += n;
    if (s.out_off == s.out_len) {
        s.out_off = 0;
        s.out_len = 0;
    }
}

pub fn hasRoom(s: *Server) bool {
    if (s.out_off != 0) {
        const n = s.output().len;
        std.mem.copyForwards(u8, s.out[0..n], s.output());
        s.out_off = 0;
        s.out_len = n;
    }
    return s.out.len - s.out_len >= @max(s.msize, msize_min);
}

/// A null result means more input or output drainage is needed.
/// Any receive error is terminal: the stream cannot be safely resynchronized.
pub fn receive(s: *Server) Error!?wire.Decoded {
    assert(s.frame == 0);
    if (s.dead) return null;
    errdefer s.dead = true;
    const len = wire.frameLen(s.in[0..s.in_len]) orelse return null;
    if (len < wire.header_len or len > s.in.len) return error.Protocol;
    if (len > s.in_len) return null;
    const got = try wire.decode(s.in[0..len]);
    const kind = got.msg.msgType();
    if (!wire.isT(kind)) return error.Protocol;
    if (!s.hasRoom()) return null;
    if (kind == .tversion) {
        if (got.tag != wire.notag) return error.Protocol;
        // Finish any partially transmitted response before starting a new session.
        if (s.output().len != 0) return null;
        s.pending = @splat(.{});
        s.versioning = true;
        s.msize = 0;
    } else {
        if (s.msize == 0 or s.versioning) return error.Protocol;
        if (len > s.msize or got.tag == wire.notag) return error.Protocol;
        if (s.find(got.tag) != null) return error.Protocol;
        const slot = s.free(kind) orelse return error.NoTags;
        slot.* = .{ .request = kind, .tag = got.tag, .count = switch (got.msg) {
            .tread => |m| m.count,
            .twrite => |m| @intCast(m.data.len),
            .twalk => |m| m.nwname,
            else => 0,
        }, .oldtag = if (got.msg == .tflush) got.msg.tflush.oldtag else wire.notag };
    }
    s.frame = len;
    return got;
}

pub fn release(s: *Server) void {
    assert(s.frame != 0 and s.frame <= s.in_len);
    const n = s.in_len - s.frame;
    std.mem.copyForwards(u8, s.in[0..n], s.in[s.frame..s.in_len]);
    s.in_len = n;
    s.frame = 0;
}

/// Call only for a received Tversion, after aborting backend work and releasing fids.
/// Suffixes may fall back to base 9P2000; arbitrary strings beginning with 9P may not.
pub fn negotiate(s: *Server, want: u32, version: []const u8) Error!void {
    assert(s.versioning);
    const size: u32 = @intCast(@min(want, s.in.len, s.out.len, std.math.maxInt(u32)));
    if (size < msize_min) return error.TooLarge;
    const base = std.mem.sliceTo(version, '.');
    const known = std.mem.eql(u8, base, "9P2000");
    try s.append(wire.notag, .{ .rversion = .{
        .msize = size,
        .version = if (known) "9P2000" else "unknown",
    } }, size);
    s.msize = if (known) size else 0;
    s.versioning = false;
}

/// An Rflush is a backend promise: no further response for oldtag will be sent.
/// A canceled backend operation must be retired before its tag can be reused.
pub fn reply(s: *Server, tag: u16, msg: wire.Msg) Error!void {
    if (s.dead) return error.Protocol;
    const slot = s.find(tag) orelse return error.UnknownTag;
    const request = slot.request.?;
    const kind = msg.msgType();
    if (request == .tflush and kind != .rflush) return error.WrongReply;
    if (kind != .rerror and @intFromEnum(kind) != @intFromEnum(request) + 1)
        return error.WrongReply;
    switch (msg) {
        .rread => |m| if (m.data.len > slot.count) return error.WrongReply,
        .rwrite => |m| if (m.count > slot.count) return error.WrongReply,
        .rwalk => |m| {
            if (m.nwqid > slot.count) return error.WrongReply;
            if (m.nwqid == 0 and slot.count != 0) return error.WrongReply;
        },
        else => {},
    }
    var response = msg;
    if (response == .rerror) {
        const cap = @min(s.msize - wire.header_len - 2, std.math.maxInt(u16));
        response.rerror.ename = response.rerror.ename[0..@min(response.rerror.ename.len, cap)];
    }
    try s.append(tag, response, s.msize);
    const oldtag = slot.oldtag;
    slot.* = .{};
    if (kind == .rflush) {
        if (s.find(oldtag)) |old| old.* = .{};
    }
}

fn append(s: *Server, tag: u16, msg: wire.Msg, limit: u32) Error!void {
    if (try wire.encodedLen(msg) > limit) return error.TooLarge;
    _ = s.hasRoom();
    const bytes = try wire.encode(msg, tag, s.out[s.out_len..]);
    s.out_len += bytes.len;
}

fn find(s: *Server, tag: u16) ?*Pending {
    for (&s.pending) |*slot| {
        if (slot.request != null and slot.tag == tag) return slot;
    }
    return null;
}

fn free(s: *Server, request: wire.Type) ?*Pending {
    for (&s.pending, 0..) |*slot, i| {
        // A full ordinary request window must still admit cancellation.
        if (i == s.pending.len - 1 and request != .tflush) continue;
        if (slot.request == null) return slot;
    }
    return null;
}

pub fn hangup(s: *Server) void {
    s.dead = true;
    s.pending = @splat(.{});
    s.in_len = 0;
    s.frame = 0;
    s.out_len = 0;
    s.out_off = 0;
    s.msize = 0;
    s.versioning = false;
}