//! The service harness (docs/process-lifecycle.md): one replyWait loop that
//! folds protocol requests, signals, and subscribed notifications into
//! callbacks — so the lifecycle contract ("answers ping, exits on terminate")
//! is satisfied by construction and a service author writes domain logic only.
//! Nothing is asynchronous inside the process: a callback runs at a point the
//! loop chose, never on a hijacked stack — the whole reason signals are
//! messages.
//!
//! One rule a service author does have to know, and it is stated on
//! `Callbacks.on_message`: **a capability that arrives belongs to the turn** —
//! the loop closes it unless the callback claims it with `take()`. Forgetting is
//! therefore safe, and keeping is explicit; the opposite arrangement quietly
//! spends a handle-table slot per request.
//!
//! The harness also owns the **subscriber side** of a protocol that declares
//! `.events` — see `Subscribers`. The table, the reserved subscribe/unsubscribe
//! verbs, the fan-out, and the dead-subscriber sweep live here rather than in
//! each provider, so every event stream in the system has identical semantics
//! (docs/os-development/protocol-namespace.md, "Wiring").
//!
//! The liveness probe: a **zero-length request is the universal ping**, answered
//! with a zero-length reply by the harness itself. No protocol's requests start
//! at length zero, so the encoding cannot collide, and there is nothing for a
//! service author to implement — a wedged service simply fails to answer, which
//! is the diagnosis (see docs/ipc.md).

const channel = @import("channel");
const envelope = @import("envelope");
const ipc = @import("ipc");
const process = @import("process");

/// The harness's handle on a provider's subscriber table, type-erased because
/// `run` is not generic over the protocol while `Subscribers` is. A service names
/// its table once, as `Callbacks.subscribers`, and the loop does the rest: it
/// subscribes to published process exits at startup and drops a dead task's
/// subscriptions before the service's own notification callback ever sees the
/// badge.
pub const SubscriberHooks = struct {
    /// Ask the kernel for published exit events on this service's endpoint.
    watch: *const fn (endpoint: ipc.Handle) void,
    /// Drop everything task `dead` had subscribed.
    forget: *const fn (dead: u32) void,
};

pub const Callbacks = struct {
    /// Called once with the service's endpoint before the loop starts — the
    /// place to subscribe to exit events, bind IRQs, or announce readiness.
    /// Return false to abort startup (the process exits).
    init: ?*const fn (endpoint: ipc.Handle) bool = null,
    /// One protocol request from `sender` (a task id): write the reply into
    /// `reply`, return its length. The zero-length ping never reaches this.
    ///
    /// `arrived` is the capability the request carried (M13 cap passing — how a
    /// subscriber hands over its endpoint), and it comes with **an ownership
    /// rule: the turn owns it, and a handler that wants to keep it must say so
    /// with `take()`.** Whatever is left when this returns, the loop closes.
    /// `peek()` reads it without claiming, which is what a handler that may
    /// still refuse wants — no close of its own on the refusal paths.
    ///
    /// The rule is stated here, in the contract, because the alternative has
    /// failed in practice: an implementation that simply ignored a `?ipc.Handle`
    /// argument leaked a handle table slot per request, and every operation
    /// except a subscribe ignores it. Thirty-two such requests — zero-length
    /// pings will do, and they need no authorization — and the service can never
    /// accept another capability for the rest of the boot. See `ipc.Arrival`.
    on_message: *const fn (message: []const u8, reply: []u8, sender: u32, arrived: *ipc.Arrival) usize,
    /// A notification that is not a signal — a subscribed exit event, a bound
    /// IRQ, a timer landing. The raw badge; decode with the ipc helpers.
    on_notification: ?*const fn (badge: u64) void = null,
    /// The reload signal. Default: ignored.
    on_reload: ?*const fn () void = null,
    /// The terminate signal, called before the loop returns. The clean exit is
    /// the return itself — never put *necessary* work here (iron rule 1: a kill
    /// arrives with no warning; this is for graceful extras only).
    on_terminate: ?*const fn () void = null,
    /// The contract this service provides: a name under `/protocol`, mirroring
    /// the `library/protocol/` module that defines the wire format — a program
    /// imports `display-protocol` and the provider binds `"display"`
    /// (docs/os-development/protocol-namespace.md). Bound at startup, before
    /// `init` runs, so the service is reachable the moment it serves. A refusal
    /// (not granted, or a live provider already holds the name) aborts startup.
    service: ?[]const u8 = null,
    /// This provider's subscriber table — `Subscribers(Protocol, Context).hooks`
    /// — for a protocol that declares `.events`. Naming it here is what buys the
    /// exit-notification sweep: the loop subscribes to published deaths at
    /// startup and releases a dead subscriber's slot (and the endpoint capability
    /// in it) when one lands.
    subscribers: ?SubscriberHooks = null,
};

/// How many subscribers one provider fans out to. Bounded like every table in
/// this system; a subscribe past the end is refused with `-ENOSPC` rather than
/// silently forgetting an earlier one.
pub const subscriber_capacity = 8;

/// The interest mask that means "every event of this protocol" — what a
/// subscriber which named no class gets, and what a provider passes when the
/// event it is publishing belongs to no class.
pub const every_event: u32 = 0;

/// The subscriber side of a protocol, for a provider whose contract declares
/// `.events` (docs/os-development/protocol-namespace.md: *the harness owns the
/// machinery — the subscriber table, the dead-subscriber sweep, and the fan-out
/// loop*). Three services hand-rolled this, with three different ideas of when a
/// dead subscriber goes away — a poll of the process list on subscribe, a drop on
/// a failed send, and nothing at all. This is the one idiom.
///
/// ```zig
/// const Subscriptions = service.Subscribers(power_protocol.Protocol, void);
/// ...
/// fn onMessage(message: []const u8, reply: []u8, sender: u32, arrived: *ipc.Arrival) usize {
///     return Subscriptions.dispatch({}, handlers, message, sender, arrived, reply);
/// }
/// pub fn main() void {
///     service.run(power_protocol.message_maximum, .{
///         .service = "power",
///         .on_message = onMessage,
///         .subscribers = Subscriptions.hooks,
///     });
/// }
/// ```
///
/// What the provider still writes is its own events — `publish(.power_button, 0,
/// .{})`. Everything else happens here: registering the caller's endpoint on the
/// reserved `subscribe` verb, taking that capability out of the turn, dropping it
/// on `unsubscribe` or on the subscriber's death, and framing one packet for the
/// whole fan-out.
///
/// The table is per instantiation (a container-level `var` inside the generic
/// type), so a process providing two contracts gets two tables and neither can
/// see the other's subscribers.
pub fn Subscribers(comptime Protocol: type, comptime Context: type) type {
    return struct {
        /// The generated dispatch this provider answers with.
        pub const Provider = Protocol.Provider(Context);
        pub const Handlers = Provider.Handlers;

        /// One registered subscriber: the endpoint events are pushed to (the
        /// capability it handed over at subscribe time, which this slot owns),
        /// the task that handed it over — the kernel-stamped badge, the only
        /// source identity there is — and which classes of event it asked for.
        const Slot = struct {
            used: bool = false,
            endpoint: ipc.Handle = 0,
            task: u32 = 0,
            interest: u32 = every_event,
        };

        var slots: [subscriber_capacity]Slot = .{Slot{}} ** subscriber_capacity;

        /// Set when a slot has taken the capability the turn carried, and read
        /// back in `dispatch`, which is where the turn's `Arrival` lives. The
        /// generated dispatch hands a handler the raw handle rather than the
        /// `Arrival` — deliberately, since a handler has no business closing the
        /// turn's property — so the *claim* has to travel back out this way. One
        /// turn, one handler, one thread: there is nothing here to race.
        var claimed = false;

        /// What `Callbacks.subscribers` is given.
        pub const hooks: SubscriberHooks = .{ .watch = watchExits, .forget = forget };

        fn watchExits(endpoint: ipc.Handle) void {
            // Published exits, not a poll of the process list: a service must
            // never depend on clients cleaning up after themselves, and it must
            // not have to walk the whole table on every subscribe to find out
            // either (docs/process-lifecycle.md, "Who learns of a death").
            _ = process.subscribeExits(endpoint);
        }

        /// Release everything task `dead` had subscribed. The slot owns the
        /// endpoint capability, so reclaiming the slot closes it — otherwise a
        /// process that subscribes and dies costs a handle-table slot that never
        /// comes back.
        pub fn forget(dead: u32) void {
            for (&slots) |*slot| {
                if (slot.used and slot.task == dead) {
                    _ = ipc.close(slot.endpoint);
                    slot.* = .{};
                }
            }
        }

        /// Whether `task` is a subscriber — the gate for an operation a provider
        /// honours from its subscribers and nobody else. The power service's
        /// shutdown is the one: the badge is kernel-stamped, so nothing in a
        /// packet can claim to be the subscriber that already ran the stop
        /// sequence.
        pub fn has(task: u32) bool {
            for (&slots) |*slot| {
                if (slot.used and slot.task == task) return true;
            }
            return false;
        }

        /// Answer one received packet, with the reserved `subscribe` and
        /// `unsubscribe` verbs already wired — a provider that leaves those two
        /// handlers null (every provider should) gets the harness's. The turn's
        /// capability is peeked, never taken, unless a slot actually kept it.
        pub fn dispatch(
            context: Context,
            handlers: Handlers,
            packet: []const u8,
            sender: u32,
            arrived: *ipc.Arrival,
            reply: []u8,
        ) usize {
            var wired = handlers;
            if (wired.subscribe == null) wired.subscribe = onSubscribe;
            if (wired.unsubscribe == null) wired.unsubscribe = onUnsubscribe;
            claimed = false;
            const written = Provider.dispatch(context, wired, packet, sender, arrived.peek(), reply);
            if (claimed) _ = arrived.take();
            return written;
        }

        /// Push one event to every subscriber.
        pub fn publish(
            comptime event: Protocol.Event,
            target: u64,
            payload: Protocol.PayloadOf(event),
        ) void {
            publishClass(event, target, payload, every_event);
        }

        /// Push one event to the subscribers whose interest mask includes
        /// `class` (a subscriber that named no class takes everything). The
        /// packet is framed **once**, outside the loop, so every subscriber of a
        /// class receives identical bytes; and delivery is `ipc.send`, which
        /// never blocks, so one slow or dead subscriber can never stall the rest
        /// — the whole reason broadcast is a provider pattern and not a kernel
        /// primitive.
        pub fn publishClass(
            comptime event: Protocol.Event,
            target: u64,
            payload: Protocol.PayloadOf(event),
            class: u32,
        ) void {
            var packet: [envelope.post_maximum]u8 = undefined;
            const framed = Protocol.encodeEvent(event, target, payload, &packet) orelse return;
            for (&slots) |*slot| {
                if (!slot.used) continue;
                if (!wants(slot.*, class)) continue;
                // The sweep is what normally reclaims a dead subscriber, promptly
                // and with its capability closed. This is the backstop for a
                // notification that never arrived: an endpoint's notify ring is
                // bounded, so a burst of deaths can drop one, and a send to an
                // endpoint whose owner is gone fails rather than blocking.
                if (!ipc.send(slot.endpoint, framed)) {
                    _ = ipc.close(slot.endpoint);
                    slot.* = .{};
                }
            }
        }

        fn wants(slot: Slot, class: u32) bool {
            if (class == every_event) return true; // the event belongs to no class
            if (slot.interest == every_event) return true; // the subscriber named none
            return slot.interest & class != 0;
        }

        /// The reserved `subscribe` verb: register the caller's endpoint (the
        /// call's capability) for the classes its tail names. A refusal simply
        /// returns and the turn closes what arrived — the harness's ownership
        /// rule (`ipc.Arrival`), which is why a subscribe storm against a full
        /// table cannot spend the handle table.
        fn onSubscribe(_: Context, invocation: envelope.Invocation(void), _: envelope.Answer(void)) isize {
            const endpoint = invocation.capability orelse return -envelope.EPROTO; // no endpoint passed
            const interest = envelope.decodeSubscribe(invocation.tail).interest;
            for (&slots) |*slot| {
                if (slot.used) continue;
                // Appended, not replaced: one task may hold several subscriptions
                // on different endpoints (a client taking keyboard and mouse as
                // two streams), and each is its own conversation.
                slot.* = .{ .used = true, .endpoint = endpoint, .task = invocation.sender, .interest = interest };
                claimed = true; // the table holds it until that task dies
                return 0;
            }
            return -envelope.ENOSPC; // table full
        }

        /// The reserved `unsubscribe` verb: every subscription the calling task
        /// holds here goes, which is exactly what its death would do. It names no
        /// endpoint because the badge already names the only subscriber a caller
        /// can speak for — its own.
        fn onUnsubscribe(_: Context, invocation: envelope.Invocation(void), _: envelope.Answer(void)) isize {
            if (!has(invocation.sender)) return -envelope.ENOENT;
            forget(invocation.sender);
            return 0;
        }
    };
}

/// Run the service: create the endpoint, bind it under the service's contract
/// name (if it has one), bind signals to it, call `init`, then serve until
/// `terminate` arrives — at which point the loop returns and main's return is
/// the clean exit the supervisor reads as `ExitReason.exited`.
/// `maximum_message` sizes the receive and reply buffers (a service passes its
/// protocol's message maximum).
pub fn run(comptime maximum_message: usize, callbacks: Callbacks) void {
    const endpoint = ipc.createIpcEndpoint() orelse return;
    if (callbacks.service) |name| {
        if (!channel.bindPatiently(name, endpoint)) return;
    }
    _ = process.bindSignals(endpoint);
    // Before `init`, so a subscriber that arrives the instant the name is bound
    // is already covered by the sweep that will release it.
    if (callbacks.subscribers) |subscribers| subscribers.watch(endpoint);
    if (callbacks.init) |initialise| {
        if (!initialise(endpoint)) return;
    }

    var reply_buffer: [maximum_message]u8 = undefined;
    var reply_len: usize = 0;
    var receive: [maximum_message]u8 = undefined;
    while (true) {
        const got = ipc.replyWait(endpoint, reply_buffer[0..reply_len], &receive, null);
        // Whatever capability came with this turn is the turn's, and the turn
        // closes it unless a callback claims it (`ipc.Arrival`). Structural
        // rather than a close per branch, because the branches are exactly what
        // gets forgotten: the ping's `continue` below, and every `on_message`
        // that has no use for a capability — which is every operation but a
        // subscribe. A `defer` in a loop body runs on `continue` and on the
        // `return` that ends the loop, so this covers all four exits.
        var arrived: ipc.Arrival = .{ .handle = got.cap };
        defer arrived.release();

        if (got.isNotification()) {
            reply_len = 0; // nothing owed for a notification
            if (process.signalsFrom(got.badge)) |signals| {
                if (signals.has(.reload)) {
                    if (callbacks.on_reload) |onReload| onReload();
                }
                if (signals.has(.terminate)) {
                    if (callbacks.on_terminate) |onTerminate| onTerminate();
                    return; // the loop's return IS the clean exit
                }
                continue;
            }
            // A death sweeps the subscriber table first, then still reaches the
            // service: a provider often has its own per-client state to release
            // (open file handles, device tokens, layers) and the same badge is
            // the notice for both.
            if (got.isChildExit()) {
                if (callbacks.subscribers) |subscribers| subscribers.forget(got.childProcessId());
            }
            if (callbacks.on_notification) |onNotification| onNotification(got.badge);
            continue;
        }
        if (got.len == 0) {
            reply_len = 0; // the universal ping: a zero-length reply, from the harness
            continue; // any capability it carried goes out through the turn's `defer`
        }
        reply_len = callbacks.on_message(receive[0..got.len], &reply_buffer, got.senderTaskId(), &arrived);
    }
}