path: root/src/ninep/events.zig

//! Event streams: per-pane `event` records, the editor-wide `log`, and the
//! bounded queues that park a read until something happens.
const std = @import("std");
const pardes = @import("../pardes.zig");
const exec = @import("../exec.zig");
const look = @import("../look.zig");
const cloud9 = @import("cloud9");
const tree = @import("tree.zig");
const pane_files = @import("pane.zig");
const dump = @import("../dump.zig");

const Pardes = pardes.Pardes;
const Pane = pardes.Pane;
const MAX_PANES = pardes.MAX_PANES;
const Req = tree.Req;
const Reply = tree.Reply;
const E = tree.E;

pub const queue_cap = 64 * 1024;

/// Wall-clock seconds, or zero where the platform has no clock.
pub fn now() u32 {
    if (comptime !pardes.hosted) return 0;
    var ts: std.c.timespec = undefined;
    if (std.c.clock_gettime(.REALTIME, &ts) != 0) return 0;
    return std.math.cast(u32, ts.sec) orelse 0;
}

pub const Queue = struct {
    buf: std.ArrayList(u8) = .empty,
    head: usize = 0,
    cap: usize = queue_cap,
    /// Records pushed out of the front to make room: the sequence number of
    /// the oldest one still held.
    dropped: u64 = 0,

    pub fn deinit(q: *Queue, gpa: std.mem.Allocator) void {
        q.buf.deinit(gpa);
        q.head = 0;
    }

    pub fn push(q: *Queue, gpa: std.mem.Allocator, record: []const u8) void {
        if (record.len > std.math.maxInt(u32)) return;
        while (q.buf.items.len - q.head + record.len + 4 > q.cap) {
            if (q.peek() == null) return;
            q.pop();
            q.dropped += 1;
        }
        q.compact();
        var head: [4]u8 = undefined;
        std.mem.writeInt(u32, &head, @intCast(record.len), .little);
        q.buf.appendSlice(gpa, &head) catch return;
        q.buf.appendSlice(gpa, record) catch {
            q.buf.shrinkRetainingCapacity(q.buf.items.len - 4);
            return;
        };
    }

    pub fn peek(q: *const Queue) ?[]const u8 {
        const rest = q.buf.items[@min(q.head, q.buf.items.len)..];
        if (rest.len < 4) return null;
        const len = std.mem.readInt(u32, rest[0..4], .little);
        if (rest.len < 4 + len) return null;
        return rest[4 .. 4 + len];
    }

    pub fn pop(q: *Queue) void {
        const record = q.peek() orelse return;
        q.head += 4 + record.len;
        if (q.head == q.buf.items.len) {
            q.buf.clearRetainingCapacity();
            q.head = 0;
        }
    }

    pub fn popFront(q: *Queue, n: usize) void {
        const record = q.peek() orelse return;
        if (n >= record.len) return q.pop();
        q.head += n;
        std.mem.writeInt(u32, q.buf.items[q.head..][0..4], @intCast(record.len - n), .little);
    }

    fn compact(q: *Queue) void {
        if (q.head == 0 or q.head * 2 < q.buf.items.len) return;
        const rest = q.buf.items.len - q.head;
        std.mem.copyForwards(u8, q.buf.items[0..rest], q.buf.items[q.head..]);
        q.buf.shrinkRetainingCapacity(rest);
        q.head = 0;
    }

    pub fn empty(q: *const Queue) bool {
        return q.peek() == null;
    }

    pub fn clearAndFree(q: *Queue, gpa: std.mem.Allocator) void {
        q.buf.clearAndFree(gpa);
        q.head = 0;
    }
};

/// One record per read; `.again` holds the read until a record arrives.
pub fn readQueue(p: *Pardes, req: Req, q: *Queue) Reply {
    const record = q.peek() orelse return .{ .tag = req.tag, .status = .again };
    if (req.size < record.len) return Reply.fail(req.tag, E.INVAL);
    const out = p.fs.stage(p.gpa);
    out.appendSlice(p.gpa, record) catch return Reply.fail(req.tag, E.NOMEM);
    q.pop();
    return .{ .tag = req.tag, .payload = .{ .staged = @intCast(out.items.len) } };
}

// ---- the editor-wide log ----

pub const LogKind = enum { new, del, rename, save };

/// A pane leaving, before its state goes with it; one never announced
/// leaves silently.
pub fn noteRetire(p: *Pardes, pane: *Pane) void {
    tree.pty.shellGone(p, pane, false);
    p.fs.news = true; // a read held on its event or pty/data hears it went
    p.fs.listeners -|= pane.fs.readers;
    if (pane.fs.unannounced) {
        pane.fs.unannounced = false;
        return;
    }
    noteLog(p, .del, pane);
}

/// Called at the end of every update: reports the panes installed by it.
pub fn announce(p: *Pardes) void {
    for (p.panes) |slot| {
        const pane = slot orelse continue;
        if (!pane.fs.unannounced) continue;
        pane.fs.unannounced = false;
        noteLog(p, .new, pane);
    }
}

/// Records `<kind> <serial> <name>`.
pub fn noteLog(p: *Pardes, kind: LogKind, pane: *Pane) void {
    var buf: [4096 + 64]u8 = undefined;
    const name = pane_files.nameOf(pane);
    pushLog(p, std.fmt.bufPrint(&buf, "{s} {d} {s}\n", .{ @tagName(kind), pane.serial, name[0..@min(name.len, 4096)] }) catch return);
}

/// Records `msg <serial> <text>` for what the editor said, `-` for no pane.
pub fn noteMessage(p: *Pardes, serial: u32, text: []const u8) void {
    var buf: [pardes.Messages.LoggedMessage.cap + 32]u8 = undefined;
    pushLog(p, (if (serial == 0)
        std.fmt.bufPrint(&buf, "msg - {s}\n", .{text})
    else
        std.fmt.bufPrint(&buf, "msg {d} {s}\n", .{ serial, text })) catch return);
}

/// Records `err <serial|-> <file>: <why>` for a write or truncation that was
/// refused or failed: through a mount a shell sees only the errno the
/// kernel mapped the reply to (`Invalid argument`), and here is the reason.
/// The log is the one place for it, as acme's `errors` file takes text and
/// answers nothing: a per-pane readable error file would be a second.
/// `dump <path>` when a Dump is written, `restore <path>` first in a
/// Restore's replacement.
pub fn notePath(p: *Pardes, what: []const u8, path: []const u8) void {
    var buf: [pardes.memory.limits.host_path_cap + 16]u8 = undefined;
    pushLog(p, std.fmt.bufPrint(&buf, "{s} {s}\n", .{ what, path }) catch return);
}

pub fn noteError(p: *Pardes, req: Req, reply: Reply) void {
    const why = if (reply.ename.len > 0) reply.ename else cloud9.fs.errString(reply.errno);
    var name: [16]u8 = undefined;
    var serial: []const u8 = "-";
    var file: []const u8 = "os";
    if (req.node & tree.sources.archive_node != 0) file = "src";
    if (req.node != pardes.filesystem.os_root and req.node & pardes.filesystem.os_node == 0 and req.node & tree.sources.archive_node == 0) {
        if (tree.Node.target(req.node)) |t| switch (t) {
            .top => |f| file = f.fileName(),
            .pane => |pt| {
                serial = std.fmt.bufPrint(&name, "{d}", .{pt.serial}) catch "-";
                file = switch (pt.file) {
                    .pty_ctl => "pty/ctl",
                    .pty_status => "pty/status",
                    .pty_data => "pty/data",
                    .pty_run => "pty/run",
                    else => pt.file.fileName(),
                };
            },
        };
    }
    var buf: [256]u8 = undefined;
    const record = std.fmt.bufPrint(&buf, "err {s} {s}: {s}\n", .{ serial, file, why[0..@min(why.len, 200)] }) catch return;
    for (record[0 .. record.len - 1]) |*c| if (c.* < ' ') {
        c.* = ' ';
    };
    // A client retrying a write that fails the same way would fill the ring
    // with one line, so a repeat of the newest record is that record counted,
    // `(x3)`, as +Messages counts its repeats; unless a follower has read it
    // already and so waits on the repeat as a line of its own.
    const q = &p.fs.log;
    var at = q.head;
    var last: ?usize = null;
    var seq = q.dropped;
    while (at + 4 <= q.buf.items.len) : (seq += 1) {
        last = at;
        at += 4 + std.mem.readInt(u32, q.buf.items[at..][0..4], .little);
    }
    const last_at = last orelse return pushLog(p, record);
    const last_seq = seq - 1;
    var said = q.buf.items[last_at + 4 ..];
    said = said[0 .. said.len - 1];
    var times: u32 = 1;
    if (std.mem.endsWith(u8, said, ")")) if (std.mem.lastIndexOf(u8, said, " (x")) |x| {
        if (std.fmt.parseInt(u32, said[x + 3 .. said.len - 1], 10)) |n| {
            times = n;
            said = said[0..x];
        } else |_| {}
    };
    if (!std.mem.eql(u8, said, record[0 .. record.len - 1])) return pushLog(p, record);
    for (p.fs.opens) |o| if (o.node != 0 and o.what == .log) {
        const l = o.what.log;
        if (l.follow and (l.next > last_seq or (l.next == last_seq and l.part > 0))) return pushLog(p, record);
    };
    var counted: [280]u8 = undefined;
    const again = std.fmt.bufPrint(&counted, "{s} (x{d})\n", .{ said, times +| 1 }) catch return;
    q.buf.shrinkRetainingCapacity(last_at);
    pushLog(p, again);
}

/// The log is one ring that records whether or not anyone reads it. A record
/// is one line: a newline in a message or a name would read as two records.
fn pushLog(p: *Pardes, record: []u8) void {
    for (record[0 .. record.len - 1]) |*c| if (c.* < ' ') {
        c.* = ' ';
    };
    // One record larger than the ring would push every other out and then
    // not fit itself; cut it to what fits instead, on a character boundary.
    var end = @min(record.len, p.fs.log.cap - 4) - 1;
    while (end > 0 and record[end] & 0xC0 == 0x80) end -= 1;
    record[end] = '\n';
    p.fs.log.push(p.gpa, record[0 .. end + 1]);
    p.fs.news = true;
}

/// What an open of /log holds.
pub const LogOpen = struct {
    /// The ring's text when it was opened.
    bytes: []const u8 = &.{},
    /// This open waits for records after `bytes`, from `next`.
    follow: bool = false,
    next: u64 = 0,
    /// How much of record `next` a short read already took.
    part: u32 = 0,
    /// How much of `bytes` this open has read. Reads walk it as a stream once
    /// following and ignore their offset, since `exec 3<>log; echo follow
    /// >&3` leaves the shared offset past the word it wrote.
    pos: u64 = 0,
};

/// An open freezes the ring's text, so `cat log` answers what happened lately
/// and ends; `next` is where a follower picks up after it.
pub fn freezeLog(p: *Pardes) !LogOpen {
    var text: std.ArrayList(u8) = .empty;
    errdefer text.deinit(p.gpa);
    var at = p.fs.log.head;
    const items = p.fs.log.buf.items;
    var n: u64 = 0;
    while (at + 4 <= items.len) : (n += 1) {
        const len = std.mem.readInt(u32, items[at..][0..4], .little);
        try text.appendSlice(p.gpa, items[at + 4 ..][0..len]);
        at += 4 + len;
    }
    return .{ .bytes = try text.toOwnedSlice(p.gpa), .next = p.fs.log.dropped + n };
}

/// Past the frozen text, a following open reads newer records one per read,
/// held until one arrives; one that fell behind the ring hears `lost N`.
pub fn readLog(p: *Pardes, req: Req) Reply {
    const slot = &(tree.openOf(p, req) orelse return Reply.fail(req.tag, E.INVAL)).what.log;
    const frozen = slot.bytes;
    const out = p.fs.stage(p.gpa);
    if (!slot.follow or slot.pos < frozen.len) {
        // Until `follow`, the frozen text is a file like any other: `tail -n`
        // seeks near its end, `less` seeks back. After it, a stream.
        const off = @min(if (slot.follow) slot.pos else req.off, frozen.len);
        const len = @min(frozen.len - off, req.size);
        out.appendSlice(p.gpa, frozen[off..][0..len]) catch return Reply.fail(req.tag, E.NOMEM);
        slot.pos = off + len;
        return .{ .tag = req.tag, .payload = .{ .staged = @intCast(len) } };
    }
    const q = &p.fs.log;
    if (slot.next < q.dropped) {
        out.print(p.gpa, "lost {d}\n", .{q.dropped - slot.next}) catch return Reply.fail(req.tag, E.NOMEM);
        slot.next = q.dropped;
        slot.part = 0;
        return .{ .tag = req.tag, .payload = .{ .staged = @intCast(out.items.len) } };
    }
    // ponytail: walks the ring from its oldest record; it holds at most log_bytes.
    var at = q.head;
    var seq = q.dropped;
    while (at + 4 <= q.buf.items.len) : (seq += 1) {
        const len = std.mem.readInt(u32, q.buf.items[at..][0..4], .little);
        if (seq == slot.next) {
            // A short read (a shell's `read` takes a few bytes at a time)
            // gets the record in pieces, as pty/data does.
            const rest = q.buf.items[at + 4 + slot.part ..][0 .. len - slot.part];
            const n = @min(rest.len, req.size);
            out.appendSlice(p.gpa, rest[0..n]) catch return Reply.fail(req.tag, E.NOMEM);
            slot.part += @intCast(n);
            if (slot.part == len) {
                slot.next += 1;
                slot.part = 0;
            }
            return .{ .tag = req.tag, .payload = .{ .staged = @intCast(n) } };
        }
        at += 4 + len;
    }
    return .{ .tag = req.tag, .status = .again };
}

/// `follow`, written to an open of the log, makes that open wait for newer
/// records instead of ending; closing it is the way back.
pub fn writeLog(p: *Pardes, req: Req) Reply {
    const slot = &(tree.openOf(p, req) orelse return Reply.fail(req.tag, E.INVAL)).what.log;
    if (!std.mem.eql(u8, std.mem.trim(u8, req.data, " \t\r\n"), "follow")) return Reply.fail(req.tag, E.INVAL);
    slot.follow = true;
    return .{ .tag = req.tag, .written = @intCast(req.data.len) };
}

/// The length a stat reports: the text an open would freeze now.
pub fn logLen(q: *const Queue) u64 {
    var n: u64 = 0;
    var at = q.head;
    while (at + 4 <= q.buf.items.len) {
        const len = std.mem.readInt(u32, q.buf.items[at..][0..4], .little);
        n += len;
        at += 4 + len;
    }
    return n;
}

// ---- per-pane event records ----

pub const max_record_text = 256;

pub const Action = enum(u8) {
    body_delete = 'D',
    tag_delete = 'd',
    body_insert = 'I',
    tag_insert = 'i',
    body_look = 'L',
    tag_look = 'l',
    body_exec = 'X',
    tag_exec = 'x',

    pub fn char(a: Action) u8 {
        return @intFromEnum(a);
    }

    pub fn fromChar(c: u8) ?Action {
        return std.enums.fromInt(Action, c);
    }

    pub fn onTag(a: Action) bool {
        return @intFromEnum(a) >= 'a';
    }
};

pub const flag_builtin: u32 = 1;
pub const flag_expansion: u32 = 2;
pub const flag_filename: u32 = 4;
pub const flag_chorded: u32 = 8;

pub fn formatRecord(
    buf: []u8,
    origin: u8,
    action: Action,
    q0: u32,
    q1: u32,
    flag: u32,
    text: []const u8,
) []const u8 {
    const sent = if (text.len >= max_record_text) text[0..0] else text;
    return std.fmt.bufPrint(buf, "{c}{c}{d} {d} {d} {d} {s}\n", .{
        origin,
        action.char(),
        q0,
        q1,
        flag,
        sent.len,
        sent,
    }) catch buf[0..0];
}

pub const Span = struct { at: u32, removed: u32, inserted: u32 };

pub fn diffSpan(old: []const u8, new: []const u8) Span {
    const both = @min(old.len, new.len);
    const stride = 64;
    var head: usize = 0;
    while (head + stride <= both and
        std.mem.eql(u8, old[head..][0..stride], new[head..][0..stride])) head += stride;
    while (head < both and old[head] == new[head]) head += 1;
    var tail: usize = 0;
    const rest = both - head;
    while (tail + stride <= rest and std.mem.eql(
        u8,
        old[old.len - tail - stride ..][0..stride],
        new[new.len - tail - stride ..][0..stride],
    )) tail += stride;
    while (tail < rest and old[old.len - 1 - tail] == new[new.len - 1 - tail]) tail += 1;
    return .{
        .at = @intCast(head),
        .removed = @intCast(old.len - tail - head),
        .inserted = @intCast(new.len - tail - head),
    };
}

pub fn noteReplace(p: *Pardes, id: usize, on_tag: bool, old: []const u8, new: []const u8) void {
    const pane = p.panes[id] orelse return;
    if (pane.fs.readers == 0) return;
    const span = diffSpan(old, new);
    if (span.removed == 0 and span.inserted == 0) return;
    if (span.removed > 0) _ = noteAction(
        p,
        id,
        if (on_tag) .tag_delete else .body_delete,
        span.at,
        span.at + span.removed,
        0,
        "",
    );
    if (span.inserted > 0) _ = noteAction(
        p,
        id,
        if (on_tag) .tag_insert else .body_insert,
        span.at,
        span.at + span.inserted,
        0,
        new[span.at..][0..span.inserted],
    );
}

pub fn noteAction(
    p: *Pardes,
    id: usize,
    action: Action,
    q0: u32,
    q1: u32,
    flag: u32,
    text: []const u8,
) bool {
    const pane = p.panes[id] orelse return false;
    if (pane.fs.readers == 0) return false;
    var buf: [max_record_text + 64]u8 = undefined;
    const record = formatRecord(&buf, p.fs.origin, action, q0, q1, flag, text);
    pane.fs.events.push(p.gpa, record);
    p.fs.news = true;
    return true;
}

pub fn notePtyOutput(p: *Pardes, id: usize, bytes: []const u8) void {
    if (id >= MAX_PANES or bytes.len == 0) return;
    const pf = &(p.panes[id] orelse return).fs;
    if (!pf.pty_reader) return;
    var off: usize = 0;
    while (off < bytes.len) {
        const n = @min(bytes.len - off, queue_cap / 2);
        pf.pty_out.push(p.gpa, bytes[off..][0..n]);
        off += n;
    }
    p.fs.news = true;
}

const EventRecord = struct { action: Action, q0: u32, q1: u32, text: ?[]const u8 = null };

const EventReader = struct {
    data: []const u8,
    i: usize = 0,

    fn next(er: *EventReader) ?EventRecord {
        if (er.i >= er.data.len) return null;
        var i = er.i;
        if (i + 2 > er.data.len) return null;
        i += 1;
        const action = Action.fromChar(er.data[i]) orelse return null;
        i += 1;
        const q0 = scanNumber(er.data, &i) orelse return null;
        const q1 = scanNumber(er.data, &i) orelse return null;
        while (i < er.data.len and er.data[i] == ' ') i += 1;
        // acme takes only `origin action q0 q1`; the record whole, as read,
        // is taken too, so a click with no place in the text (one written
        // to a pane's own look or exec) can be written back and done.
        var text: ?[]const u8 = null;
        if (i < er.data.len and er.data[i] != '\n') {
            _ = scanNumber(er.data, &i) orelse return null;
            const n = scanNumber(er.data, &i) orelse return null;
            if (i >= er.data.len or er.data[i] != ' ' or er.data.len - (i + 1) < n) return null;
            text = er.data[i + 1 ..][0..n];
            i += 1 + n;
        }
        if (i >= er.data.len or er.data[i] != '\n') return null;
        er.i = i + 1;
        return .{ .action = action, .q0 = q0, .q1 = q1, .text = text };
    }
};

fn scanNumber(data: []const u8, i: *usize) ?u32 {
    while (i.* < data.len and data[i.*] == ' ') i.* += 1;
    const s = i.*;
    var n: u64 = 0;
    while (i.* < data.len and data[i.*] >= '0' and data[i.*] <= '9') : (i.* += 1)
        n = @min(n * 10 + (data[i.*] - '0'), std.math.maxInt(u32));
    if (i.* == s) return null;
    return @intCast(n);
}

/// Writing a Look or Exec record back performs the action it names.
pub fn writeEvent(p: *Pardes, req: Req, id: usize) Reply {
    const pane0 = p.panes[id] orelse return Reply.fail(req.tag, E.NOENT);
    const serial = pane0.serial;
    {
        const body = pane_files.bodyOf(pane0);
        const tag = pane_files.tagOf(p, pane0);
        var check: EventReader = .{ .data = req.data };
        while (check.next()) |r| {
            switch (r.action) {
                .body_look, .tag_look, .body_exec, .tag_exec => {},
                else => return tree.failText(req.tag, E.INVAL, tree.e_bad_event),
            }
            const n = if (r.action.onTag()) tag.len else body.len;
            if (r.q0 > r.q1 or r.q1 > n) return tree.failText(req.tag, E.INVAL, tree.e_bad_event);
        }
        if (check.i != req.data.len) return tree.failText(req.tag, E.INVAL, tree.e_bad_event);
    }
    var run: EventReader = .{ .data = req.data };
    while (run.next()) |r| {
        const live = p.paneBySerial(serial) orelse break;
        const pane = p.panes[live].?;
        const whole = if (r.action.onTag()) pane_files.tagOf(p, pane) else pane_files.bodyOf(pane);
        const lo = @min(@as(usize, r.q0), whole.len);
        const hi = @max(lo, @min(@as(usize, r.q1), whole.len));
        const said = if (r.q0 == r.q1) r.text orelse whole[lo..hi] else whole[lo..hi];
        const text = p.scratch.allocator().dupe(u8, said) catch continue;
        switch (r.action) {
            .body_exec, .tag_exec => _ = exec.execute(p, live, text),
            .body_look, .tag_look => look.lookAt(p, live, text),
            else => unreachable,
        }
    }
    return .{ .tag = req.tag, .written = @intCast(req.data.len) };
}

// ---- tests ----

const testing = std.testing;
const th = @import("testing.zig");
const call = th.call;
const rd = th.rd;
const wr = th.wr;
const withFile = th.withFile;
const serialOf = th.serialOf;
const Node = tree.Node;
const Status = tree.Status;

test "event records are acme's bytes, one per read, and .again when empty" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "Msg fs-ran\n");
    defer p.deinit();
    const serial = serialOf(p);
    const event = Node.of(serial, .event);

    _ = noteAction(p, 0, .body_exec, 1, 4, flag_builtin, "sg ");
    try testing.expect(p.panes[0].?.fs.events.empty());

    const h = call(p, .{ .tag = 10, .op = .open, .node = event });
    try testing.expect(h.reply.handle != 0);
    try testing.expectEqual(@as(u16, 1), p.fs.listeners);

    try testing.expectEqual(Status.again, rd(p, event, 0, 4096).reply.status);

    p.fs.origin = 'M';
    _ = noteAction(p, 0, .body_exec, 1, 4, flag_builtin, "ell");
    _ = noteAction(p, 0, .body_delete, 0, 3, 0, "");
    try testing.expectEqualStrings("MX1 4 1 3 ell\n", rd(p, event, 0, 4096).bytes);
    try testing.expectEqualStrings("MD0 3 0 0 \n", rd(p, event, 0, 4096).bytes);
    try testing.expectEqual(Status.again, rd(p, event, 0, 4096).reply.status);

    _ = noteAction(p, 0, .body_look, 0, 3, flag_filename, "one");
    try testing.expectEqual(E.INVAL, rd(p, event, 0, 4).errno());
    try testing.expectEqualStrings("ML0 3 4 3 one\n", rd(p, event, 0, 4096).bytes);

    const big = "z" ** max_record_text;
    _ = noteAction(p, 0, .body_exec, 0, max_record_text, 0, big);
    try testing.expectEqualStrings("MX0 256 0 0 \n", rd(p, event, 0, 4096).bytes);

    _ = call(p, .{ .tag = 11, .op = .release, .node = event, .handle = h.reply.handle });
    try testing.expectEqual(@as(u16, 0), p.fs.listeners);
}

fn drainEvents(p: *Pardes, node: u64, store: []u8, out: [][]const u8) [][]const u8 {
    var used: usize = 0;
    var n: usize = 0;
    while (n < out.len) {
        const a = rd(p, node, 0, 4096);
        if (a.reply.status != .ok) break;
        @memcpy(store[used..][0..a.bytes.len], a.bytes);
        out[n] = store[used..][0..a.bytes.len];
        used += a.bytes.len;
        n += 1;
    }
    return out[0..n];
}

test "a write through the filesystem is reported once, attributed to the file it came through" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "one\ntwo\n");
    defer p.deinit();
    const serial = serialOf(p);
    const event = Node.of(serial, .event);
    _ = call(p, .{ .tag = 40, .op = .open, .node = event });
    var store: [4096]u8 = undefined;
    var slots: [16][]const u8 = undefined;
    _ = drainEvents(p, event, &store, &slots);

    _ = wr(p, Node.of(serial, .body), "three\n");
    const body_recs = drainEvents(p, event, &store, &slots);
    try testing.expect(body_recs.len >= 1);
    try testing.expectEqualStrings("EI8 14 0 6 three\n\n", body_recs[0]);
    for (body_recs[1..]) |r| {
        try testing.expectEqual(@as(u8, 'E'), r[0]);
        try testing.expect(Action.fromChar(r[1]).?.onTag());
    }

    _ = wr(p, Node.of(serial, .addr), "1");
    _ = wr(p, Node.of(serial, .data), "ONE\n");
    const data_recs = drainEvents(p, event, &store, &slots);
    try testing.expectEqual(@as(usize, 2), data_recs.len);
    try testing.expectEqualStrings("FD0 3 0 0 \n", data_recs[0]);
    try testing.expectEqualStrings("FI0 3 0 3 ONE\n", data_recs[1]);
    try testing.expectEqual(Status.again, rd(p, event, 0, 4096).reply.status);
}

test "one event reader at a time, and a writer still holds the pane" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "x\n");
    defer p.deinit();
    const event = Node.of(serialOf(p), .event);

    const reader = call(p, .{ .tag = 12, .op = .open, .node = event });
    try testing.expectEqual(Status.ok, reader.reply.status);
    // Reads consume, so a second reader would see half the clicks: refused.
    const second = call(p, .{ .tag = 13, .op = .open, .node = event, .omode = 2 });
    try testing.expectEqual(Status.err, second.reply.status);
    try testing.expectEqualStrings(tree.e_in_use, second.reply.ename);
    const writer = call(p, .{ .tag = 13, .op = .open, .node = event, .omode = 1 });
    try testing.expectEqual(Status.ok, writer.reply.status);
    try testing.expectEqual(@as(u16, 2), p.panes[0].?.fs.readers);
    try testing.expectEqual(@as(u16, 2), p.fs.listeners);

    _ = call(p, .{ .tag = 14, .op = .release, .node = event, .handle = writer.reply.handle });
    try testing.expectEqual(@as(u16, 1), p.panes[0].?.fs.readers);
    try testing.expect(p.panes[0].?.fs.readers != 0);

    _ = call(p, .{ .tag = 15, .op = .release, .node = Node.of(serialOf(p), .body) });
    try testing.expectEqual(@as(u16, 1), p.fs.listeners);

    _ = call(p, .{ .tag = 16, .op = .release, .node = event, .handle = reader.reply.handle });
    try testing.expectEqual(@as(u16, 0), p.fs.listeners);
    try testing.expect(p.panes[0].?.fs.readers == 0);
    _ = call(p, .{ .tag = 17, .op = .release, .node = event });
    try testing.expectEqual(@as(u16, 0), p.fs.listeners);
    // Once it closed, the next reader gets in.
    const next = call(p, .{ .tag = 18, .op = .open, .node = event });
    try testing.expectEqual(Status.ok, next.reply.status);
    _ = call(p, .{ .tag = 19, .op = .release, .node = event, .handle = next.reply.handle });
}

test "a pane deleted while its event file is open leaves no suppression behind" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "x\n");
    defer p.deinit();
    const serial = serialOf(p);
    const event = Node.of(serial, .event);
    _ = try th.newPane(p);

    const a = call(p, .{ .tag = 18, .op = .open, .node = event });
    const b = call(p, .{ .tag = 19, .op = .open, .node = event, .omode = 1 });
    try testing.expectEqual(@as(u16, 2), p.fs.listeners);

    // Through ctl: a line written to its exec would go to the reader.
    _ = wr(p, Node.of(serial, .ctl), "Del\n");
    try testing.expect(p.paneBySerial(serial) == null);
    try testing.expectEqual(@as(u16, 0), p.fs.listeners);
    // The reader's next read hears what acme says of a window gone under it.
    const shut = call(p, .{ .tag = 19, .op = .read, .node = event, .handle = a.reply.handle, .size = 64 });
    try testing.expectEqual(E.IO, shut.errno());
    try testing.expectEqualStrings(tree.e_shut_down, shut.reply.ename);

    _ = call(p, .{ .tag = 20, .op = .release, .node = event, .handle = a.reply.handle });
    _ = call(p, .{ .tag = 21, .op = .release, .node = event, .handle = b.reply.handle });
    try testing.expectEqual(@as(u16, 0), p.fs.listeners);

    try testing.expectEqual(E.NOENT, rd(p, event, 0, 64).errno());
    try testing.expectEqual(E.NOENT, rd(p, Node.of(serial, .body), 0, 64).errno());
    try testing.expectEqual(E.NOENT, wr(p, Node.of(serial, .dirty), "0\n").errno());
    try testing.expectEqual(E.NOENT, call(p, .{ .tag = 22, .op = .open, .node = event }).errno());
}

test "writing an event record back performs the action it names" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "Msg fs-ran\n");
    defer p.deinit();
    const serial = serialOf(p);
    const event = Node.of(serial, .event);
    const pane = p.panes[0].?;

    const w = wr(p, event, "FX0 10\n");
    try testing.expectEqual(Status.ok, w.reply.status);
    try testing.expectEqualStrings("fs-ran", pane.msg[0..pane.msg_len]);

    pane.msg_len = 0;
    try testing.expectEqual(Status.ok, wr(p, event, "FX0 10\nFX0 10\n").reply.status);
    try testing.expectEqualStrings("fs-ran", pane.msg[0..pane.msg_len]);

    pane.msg_len = 0;
    for ([_][]const u8{
        "FX0 10\nFQ0 1\n", // unknown type character
        "FX0 999\n", // out of range
        "FX0 10", // no newline
        "FX5 1\n", // q0 > q1
        "FD0 3\n", // a report, not a request
        "F\n",
    }) |bad| {
        try testing.expectEqual(E.INVAL, wr(p, event, bad).errno());
        try testing.expectEqual(@as(usize, 0), pane.msg_len);
    }

    _ = call(p, .{ .tag = 23, .op = .open, .node = event });
    p.fs.origin = 'K';
    _ = wr(p, event, "KX0 10\n");
    try testing.expectEqual(@as(u8, 'F'), p.fs.origin);
}

test "the log records whether or not anyone reads, and an open that follows waits" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "logged\n");
    defer p.deinit();
    const log = @intFromEnum(tree.TopFile.log);
    var expected: [4200]u8 = undefined;

    // Nobody holds /log open, and the pane is recorded anyway.
    const early = try th.newPane(p);
    const early_line = try std.fmt.bufPrint(&expected, "new {d} {s}\n", .{ early, pane_files.nameOf(p.panes[p.paneBySerial(early).?].?) });
    const early_owned = try gpa.dupe(u8, early_line);
    defer gpa.free(early_owned);

    // A plain open answers what happened so far, and ends: `cat log` never hangs.
    const plain = call(p, .{ .tag = 1, .op = .open, .node = log });
    try testing.expectEqual(Status.ok, plain.reply.status);
    const first = call(p, .{ .tag = 2, .op = .read, .node = log, .handle = plain.reply.handle, .size = 8192 });
    try testing.expect(std.mem.endsWith(u8, first.bytes, early_owned));
    const past = call(p, .{ .tag = 3, .op = .read, .node = log, .handle = plain.reply.handle, .off = first.bytes.len, .size = 8192 });
    try testing.expectEqual(Status.ok, past.reply.status);
    try testing.expectEqual(@as(usize, 0), past.bytes.len);
    const tail = call(p, .{ .tag = 3, .op = .read, .node = log, .handle = plain.reply.handle, .off = first.bytes.len - early_owned.len, .size = 8192 });
    try testing.expectEqualStrings(early_owned, tail.bytes);
    _ = call(p, .{ .tag = 4, .op = .release, .node = log, .handle = plain.reply.handle });

    // `follow` on an open makes it wait past what it froze, one record a read.
    const f = call(p, .{ .tag = 5, .op = .open, .node = log });
    const fh = f.reply.handle;
    const frozen = call(p, .{ .tag = 6, .op = .read, .node = log, .handle = fh, .size = 8192 }).bytes.len;
    try testing.expectEqual(Status.err, call(p, .{ .tag = 7, .op = .write, .node = log, .handle = fh, .data = "tail\n" }).reply.status);
    try testing.expectEqual(Status.ok, call(p, .{ .tag = 8, .op = .write, .node = log, .handle = fh, .data = "follow\n" }).reply.status);
    // A stream: after `echo follow >&3` a shell reads at offset 7, and gets
    // what comes next, not the frozen text from byte 7. Before `follow` the
    // offset is honoured, so `tail -n` works (checked below on `plain`).
    const rdf = struct {
        fn next(pp: *Pardes, node: u64, h: u32, off: u64) th.Answer {
            return call(pp, .{ .tag = 9, .op = .read, .node = node, .handle = h, .off = off, .size = 8192 });
        }
    };
    // The word it refused is on record, and then there is nothing new.
    try testing.expect(std.mem.startsWith(u8, rdf.next(p, log, fh, frozen).bytes, "err - log: "));
    try testing.expectEqual(Status.again, rdf.next(p, log, fh, frozen).reply.status);

    const serial = try th.newPane(p);
    const id = p.paneBySerial(serial).?;
    try testing.expectEqualStrings(
        try std.fmt.bufPrint(&expected, "new {d} {s}\n", .{ serial, pane_files.nameOf(p.panes[id].?) }),
        rdf.next(p, log, fh, frozen).bytes,
    );
    try testing.expectEqual(Status.ok, wr(p, Node.of(serial, .name), "/tmp/logged.txt\n").reply.status);
    try testing.expectEqualStrings(
        try std.fmt.bufPrint(&expected, "rename {d} /tmp/logged.txt\n", .{serial}),
        rdf.next(p, log, fh, frozen).bytes,
    );
    // What the editor says is in the same stream, one line even when the
    // message had several.
    p.setMessage(id, "save: two\nlines");
    try testing.expectEqualStrings(
        try std.fmt.bufPrint(&expected, "msg {d} save: two lines\n", .{serial}),
        rdf.next(p, log, fh, frozen).bytes,
    );
    // A builtin announces itself on the message row, so the log says which ran.
    try testing.expectEqual(Status.ok, wr(p, Node.of(serial, .exec), "Del\n").reply.status);
    try testing.expectEqualStrings(
        try std.fmt.bufPrint(&expected, "msg {d} Del\n", .{serial}),
        rdf.next(p, log, fh, frozen).bytes,
    );
    try testing.expectEqualStrings(
        try std.fmt.bufPrint(&expected, "del {d} /tmp/logged.txt\n", .{serial}),
        rdf.next(p, log, fh, frozen).bytes,
    );
    try testing.expectEqual(Status.again, rdf.next(p, log, fh, frozen).reply.status);

    // A shell's `read` asks for a few bytes at a time: the record comes in
    // pieces rather than failing. Every repeat is its own line.
    const said_by = p.panes[0].?.serial;
    p.setMessage(0, "same failure");
    p.setMessage(0, "same failure");
    var pieced: [64]u8 = undefined;
    var got: usize = 0;
    while (got == 0 or pieced[got - 1] != '\n') {
        const piece = call(p, .{ .tag = 9, .op = .read, .node = log, .handle = fh, .off = frozen, .size = 3 }).bytes;
        @memcpy(pieced[got..][0..piece.len], piece);
        got += piece.len;
    }
    try testing.expectEqualStrings(try std.fmt.bufPrint(&expected, "msg {d} same failure\n", .{said_by}), pieced[0..got]);
    try testing.expectEqualStrings(try std.fmt.bufPrint(&expected, "msg {d} same failure\n", .{said_by}), rdf.next(p, log, fh, frozen).bytes);

    // A follower the ring outran hears how much it missed, then carries on.
    var filler: [200]u8 = @splat('x');
    for (0..p.fs.log.cap / filler.len + 8) |i| {
        _ = std.fmt.bufPrint(&filler, "{d:0>6}", .{i}) catch unreachable; // a repeat would collapse
        p.setMessage(0, &filler);
    }
    const lost = rdf.next(p, log, fh, frozen).bytes;
    try testing.expect(std.mem.startsWith(u8, lost, "lost "));
    try testing.expect(std.mem.startsWith(u8, rdf.next(p, log, fh, frozen).bytes, "msg "));
    _ = call(p, .{ .tag = 10, .op = .release, .node = log, .handle = fh });
    for (p.fs.opens) |o| try testing.expect(o.node == 0);

    // A record bigger than the whole ring is cut to fit, not dropped with
    // everything else pushed out ahead of it.
    p.fs.log.cap = 128;
    var huge: [300]u8 = @splat('y');
    p.setMessage(0, &huge);
    const cut = call(p, .{ .tag = 11, .op = .open, .node = log });
    const kept = call(p, .{ .tag = 12, .op = .read, .node = log, .handle = cut.reply.handle, .size = 512 }).bytes;
    try testing.expectEqual(@as(usize, 124), kept.len);
    try testing.expect(std.mem.startsWith(u8, kept, "msg ") and kept[kept.len - 1] == '\n');
    _ = call(p, .{ .tag = 13, .op = .release, .node = log, .handle = cut.reply.handle });
}

test "a refused or failed write is an err record in the log, saying which file and why" {
    const p = try withFile(testing.allocator, "one\n");
    defer p.deinit();
    p.update(.tick); // the pane's `new` is logged before the log is opened
    while (p.nextEffect()) |_| {}
    const serial = serialOf(p);
    const log = @intFromEnum(tree.TopFile.log);
    const f = call(p, .{ .tag = 1, .op = .open, .node = log }).reply.handle;
    const frozen = call(p, .{ .tag = 2, .op = .read, .node = log, .handle = f, .size = 1 << 16 }).bytes.len;
    _ = call(p, .{ .tag = 3, .op = .write, .node = log, .handle = f, .data = "follow" });
    var want: [128]u8 = undefined;
    const next = struct {
        fn record(pp: *Pardes, h: u32, off: usize) []const u8 {
            return call(pp, .{ .tag = 4, .op = .read, .node = @intFromEnum(tree.TopFile.log), .handle = h, .off = off, .size = 4096 }).bytes;
        }
    };
    _ = wr(p, Node.of(serial, .addr), "/nowhere/");
    try testing.expectEqualStrings(try std.fmt.bufPrint(&want, "err {d} addr: no match for regexp\n", .{serial}), next.record(p, f, frozen));
    _ = wr(p, @intFromEnum(tree.TopFile.ctl), "Bogus");
    try testing.expectEqualStrings("err - ctl: unknown control message \"Bogus\"\n", next.record(p, f, frozen));
    _ = wr(p, @intFromEnum(tree.TopFile.focus), "99999");
    try testing.expectEqualStrings("err - focus: no such window\n", next.record(p, f, frozen));
    // A write that succeeds says nothing.
    _ = wr(p, Node.of(serial, .addr), "#0");
    try testing.expectEqual(Status.again, call(p, .{ .tag = 5, .op = .read, .node = log, .handle = f, .off = frozen, .size = 4096 }).reply.status);
    // A follower that has read an err waits on its repeat: a line of its own.
    _ = wr(p, Node.of(serial, .addr), "/nowhere/");
    try testing.expectEqualStrings(try std.fmt.bufPrint(&want, "err {d} addr: no match for regexp\n", .{serial}), next.record(p, f, frozen));
    _ = wr(p, Node.of(serial, .addr), "/nowhere/");
    try testing.expectEqualStrings(try std.fmt.bufPrint(&want, "err {d} addr: no match for regexp\n", .{serial}), next.record(p, f, frozen));
    _ = call(p, .{ .tag = 6, .op = .release, .node = log, .handle = f });

    // With the follower gone, repeats are the newest record counted: the
    // one it read, and three more.
    for (0..3) |_| _ = wr(p, Node.of(serial, .addr), "/elsewhere/");
    const g = call(p, .{ .tag = 7, .op = .open, .node = log }).reply.handle;
    const text = call(p, .{ .tag = 8, .op = .read, .node = log, .handle = g, .size = 1 << 16 }).bytes;
    try testing.expect(std.mem.endsWith(u8, text, try std.fmt.bufPrint(&want, "regexp\nerr {d} addr: no match for regexp (x4)\n", .{serial})));
    _ = call(p, .{ .tag = 9, .op = .release, .node = log, .handle = g });
}

test "a Dump written and a Restore made are in the log, with their files" {
    const p = try withFile(testing.allocator, "one\n");
    defer p.deinit();
    p.setLastDump("/tmp/pardes.dump.zon");
    try dump.dumpState(p);
    const restored = try dump.restore(p, p.dump_out.?, "/tmp/pardes.dump.zon");
    defer restored.deinit();
    for ([_]*Pardes{ p, restored }, [_][]const u8{ "dump", "restore" }) |core, what| {
        const log = try freezeLog(core);
        defer core.gpa.free(log.bytes);
        var want: [64]u8 = undefined;
        try testing.expect(std.mem.endsWith(u8, log.bytes, try std.fmt.bufPrint(&want, "{s} /tmp/pardes.dump.zon\n", .{what})));
    }
}

test "opens of the log share the open records, and a closed one frees its record" {
    const gpa = testing.allocator;
    const p = try withFile(gpa, "x\n");
    defer p.deinit();
    const log = @intFromEnum(tree.TopFile.log);
    var handles: [tree.open_slots]u32 = undefined;
    for (&handles) |*h| h.* = call(p, .{ .tag = 1, .op = .open, .node = log }).reply.handle;
    try testing.expectEqual(E.NFILE, call(p, .{ .tag = 2, .op = .open, .node = log }).errno());
    _ = call(p, .{ .tag = 3, .op = .release, .node = log, .handle = handles[5] });
    const again = call(p, .{ .tag = 4, .op = .open, .node = log });
    try testing.expectEqual(Status.ok, again.reply.status);
    handles[5] = again.reply.handle;
    for (handles) |h| _ = call(p, .{ .tag = 5, .op = .release, .node = log, .handle = h });
    for (p.fs.opens) |o| try testing.expect(o.node == 0);
}

test "a click in the body reports the offsets of the word it takes, so the record can be written back" {
    const config = @import("../config.zig");
    const gpa = testing.allocator;
    const text = "alpha qty beta\n    qty\n\tqty\ndef f(qty): return qty.x\n" ++ "w" ** 100 ++ " qty\n";
    const p = try withFile(gpa, text);
    defer p.deinit();
    // A scratch made over 9P, clicked while another pane has the keyboard.
    const scratch = try th.newPane(p);
    _ = wr(p, Node.of(scratch, .body), text);
    for ([_]u32{ serialOf(p), scratch }) |serial| {
        const id = p.paneBySerial(serial).?;
        p.active = p.paneBySerial(if (serial == scratch) serialOf(p) else scratch).?;
        const event = Node.of(serial, .event);
        const h = call(p, .{ .tag = 1, .op = .open, .node = event }).reply.handle;
        var frame = std.heap.ArenaAllocator.init(gpa);
        defer frame.deinit();
        _ = try p.render(frame.allocator());
        const rect = p.rects[id];
        const x0 = rect.x + config.GUTTER + config.PREFIX_W;
        const y0 = if (p.settings.tag_bottom) rect.y else rect.y + p.panes[id].?.tag_rows;
        const tab: u16 = config.tab_width;
        const Click = struct { x: u16, y: u16, look: []const u8, exec: []const u8, scroll: i32 = 0 };
        // Where the wrapped line's `qty` landed, as drawn.
        const surface = try p.render(frame.allocator());
        var wrap_col: u16 = 0;
        while (wrap_col < rect.w and !std.mem.eql(u8, surface.at(x0 + wrap_col, y0 + 5).grapheme(), "q")) wrap_col += 1;
        for ([_]Click{
            .{ .x = x0 + 7, .y = y0, .look = "ML6 9 0 3 qty\n", .exec = "MX6 9 0 3 qty\n" },
            .{ .x = x0 + 5, .y = y0 + 1, .look = "ML19 22 0 3 qty\n", .exec = "MX19 22 0 3 qty\n" },
            .{ .x = x0 + tab + 1, .y = y0 + 2, .look = "ML24 27 0 3 qty\n", .exec = "MX24 27 0 3 qty\n" },
            .{ .x = x0 + 7, .y = y0 + 3, .look = "ML34 37 0 3 qty\n", .exec = "MX34 37 0 3 qty\n" },
            // Scrolled a line, the first row shows the second.
            .{ .x = x0 + 5, .y = y0, .look = "ML19 22 0 3 qty\n", .exec = "MX19 22 0 3 qty\n", .scroll = 1 },
            // Wrapped: the word on the continuation row.
            .{ .x = x0 + wrap_col, .y = y0 + 5, .look = "ML154 157 0 3 qty\n", .exec = "MX154 157 0 3 qty\n" },
        }) |c| for ([_]pardes.Mouse.Button{ config.look_button, config.exec_button }, [_][]const u8{ c.look, c.exec }) |button, want| {
            // The pointer rests on the word first, as a hand's does, and
            // the hover preview comes up under it.
            p.panes[id].?.scrollBy(c.scroll);
            _ = try p.render(frame.allocator());
            p.update(.{ .mouse = .{ .button = .none, .kind = .motion, .col = c.x, .row = c.y } });
            for (0..200) |_| p.update(.tick);
            p.update(.{ .mouse = .{ .button = button, .kind = .press, .col = c.x, .row = c.y } });
            p.update(.{ .mouse = .{ .button = button, .kind = .release, .col = c.x, .row = c.y } });
            try testing.expectEqualStrings(want, call(p, .{ .tag = 2, .op = .read, .node = event, .handle = h, .size = 256 }).bytes);
            p.panes[id].?.scrollBy(-c.scroll);
        };
        // A click on nothing expands to nothing, and acme then says
        // nothing (look.c:37, exec.c:141).
        p.update(.{ .mouse = .{ .button = config.look_button, .kind = .press, .col = x0 + 60, .row = y0 + 1 } });
        p.update(.{ .mouse = .{ .button = config.look_button, .kind = .release, .col = x0 + 60, .row = y0 + 1 } });
        try testing.expectEqual(Status.again, call(p, .{ .tag = 2, .op = .read, .node = event, .handle = h, .size = 256 }).reply.status);
        _ = call(p, .{ .tag = 3, .op = .release, .node = event, .handle = h });
    }
}