gunnar-grpc 1.5.0

//! `Events.Watch`, the server-streaming feed (the event stream).
//!
//! Almost the whole implementation is "build a filter, subscribe to the bus,
//! box the stream". Everything interesting (bounded memory, the gap marker for
//! a lagging subscriber, the `StreamOpened` prologue and the `from_seq` replay)
//! lives in [`crate::bus`], because that behaviour has to be testable without a
//! gRPC connection in the way.
//!
//! # The exception: a stream outlives the request that opened it
//!
//! Every other RPC in the control plane is a request, a check and an answer, so
//! "authorise in the handler" is the whole of it. This one is a request, a
//! check, and then an open pipe that keeps delivering for as long as the client
//! keeps reading — hours, in the console's case. `caller`/`require` ran once,
//! when the pipe was opened, and after that the stream carried every event on
//! the server to whoever had once held a server-wide `Read`. Revoking the key,
//! deleting the principal, dropping the grant: all of them took effect for the
//! NEXT `Watch` and for none of the open ones.
//!
//! Nothing gunnar had would close it either. `Watch` is server-scoped on
//! purpose (see the handler), so this is the broadest read in the control plane
//! — pushes, deletions, repository names across every namespace, including the
//! ones the ex-subscriber never had access to.
//!
//! [`Revalidated`] is the answer, and the property it enforces is about DATA
//! rather than about time: **no event is handed to a client across a check
//! older than [`RECHECK`]**. See its own doc for why that is the shape rather
//! than a timer.

use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use gunnar_proto::events_server::Events;
use gunnar_proto::{Event, WatchRequest};
use tokio_stream::Stream;
use tonic::{Request, Response, Status};

use crate::auth::{Access, Caller};
use crate::bus::EventFilter;
use crate::service::{EventsService, ServiceContext};
use gunnar_proto::GrantLevel;

/// **How stale the authorisation behind a delivered event may be.**
///
/// Thirty seconds. The cost of the check is one ACL read against a control
/// plane the same process is already reading on every RPC, and it is paid at
/// most twice a minute per subscriber, so the number could be smaller; it is
/// not, because a revocation that takes half a minute to reach an open pipe is
/// indistinguishable from one that takes no time at all to anybody who is not
/// holding a stopwatch, and a per-event check on a hot bus is a read per event.
pub const RECHECK: Duration = Duration::from_secs(30);

/// A `Watch` stream that keeps asking whether its subscriber may still watch.
///
/// # Why the check hangs off delivery and not off a timer
///
/// A timer would need a wakeup of its own, and would then be racing the thing
/// it guards: a revoked subscriber on a busy bus could be handed a hundred
/// events between the revocation and the tick. Checking on the delivery path
/// instead makes the guarantee a statement about what crosses the wire, which
/// is what the finding is about — **no event reaches the client unless an
/// authorisation no older than [`RECHECK`] said it could**. A quiet stream is
/// not re-checked and does not need to be: it is delivering nothing.
///
/// A failed check ends the stream, it does not skip an event. A subscriber
/// whose grant is gone must be told, in the status it was refused with, rather
/// than left holding a pipe that has silently stopped carrying — which is what
/// `EventFilter` validation up in the handler already refuses to do for a
/// mistyped namespace, for the same reason.
struct Revalidated {
    inner: crate::bus::EventStream,
    ctx: ServiceContext,
    caller: Caller,
    /// What this subscriber was let in on, re-asked at every check.
    scope: WatchScope,
    /// When the next delivery must re-check. Set to now at construction, so the
    /// FIRST event out of a long replay is covered too.
    next_check: Instant,
    /// [`RECHECK`], as a field so the constant lives in one place and a test
    /// can ask for a check on every delivery rather than sleeping for half a
    /// minute to observe one.
    recheck: Duration,
    /// Set once the check has failed. A `Stream` must not be polled for more
    /// after it has yielded its terminal error, and the refusal is terminal.
    refused: bool,
}

/// **What a watcher was let in on** (owner, 2026-10-03: an account's emitted
/// events are not public by default, "such things has to be set").
#[derive(Debug, Clone, PartialEq, Eq)]
enum WatchScope {
    /// A server reader: every event its filter lets through, as always.
    Server,
    /// Named accounts and nothing else — the request's `namespaces`, which for
    /// this watcher are a BOUNDARY, not a filter. `outside` are the ones it is
    /// not inside (watched because their events are public): an event about a
    /// repository there reaches it only when it may read that repository, so a
    /// private repository's events never leave the account.
    Accounts {
        names: Vec<String>,
        outside: Vec<String>,
    },
    /// **The caller's own** (2026-10-04, the thin layer): a non-server reader
    /// that named NO account. It cannot name the accounts it is inside — that
    /// is gunnar's to know — so the boundary is decided per EVENT: an event
    /// about a repository reaches it when the caller is inside that account
    /// (`Read` over its namespace, gates honoured), or when the account's
    /// events are public and the caller may read that repository. The
    /// stream's own bookkeeping (`StreamOpened`, `EventGap`) reaches it; an
    /// event about no repository reaches it only when it is a `KeyUsed` of the
    /// caller's own principal.
    Mine,
}

/// The one sentence for an account whose events are not shown to this caller —
/// the same whether the account exists or not, so a refusal tells an outsider
/// nothing about which names are taken.
fn not_shown(account: &str) -> Status {
    Status::permission_denied(format!(
        "Events.Watch: the events of {account} are shown only inside the account, unless its \
         account admin makes them public"
    ))
}

impl WatchScope {
    /// Decide (or re-decide) a watch. `Server` when the caller holds a
    /// server-wide `Read`; otherwise every named account must be one the caller
    /// is inside (`Read` over its namespace) or one whose events are public.
    fn decide(ctx: &ServiceContext, caller: &Caller, namespaces: &[String]) -> Result<WatchScope, Status> {
        let server = ctx.require(caller, Access::Server, GrantLevel::Read, "Events.Watch");
        match server {
            Ok(_) => return Ok(WatchScope::Server),
            Err(status) if status.code() != tonic::Code::PermissionDenied => return Err(status),
            // Naming no account: the caller's own. Let in on the caller's own
            // account (`Read` over the namespace named after it) — which is
            // true for any principal that exists and is not locked out — so a
            // removed or gated principal is refused here, in gunnar's words,
            // and the same question is the re-check below. A MEMBER has no
            // account of his own (2026-10-07): his is the account that
            // enrolled him, read on the `Read` enrolment gave him.
            Err(_) if namespaces.is_empty() => {
                let own = ctx.employer_of(&caller.principal_id)?.unwrap_or_else(|| caller.principal_id.clone());
                ctx.require(caller, Access::namespace(&own), GrantLevel::Read, "Events.Watch")?;
                return Ok(WatchScope::Mine);
            }
            Err(_) => {}
        }
        let mut outside = Vec::new();
        for account in namespaces {
            let inside = ctx
                .control
                .may(&caller.principal_id, &Access::namespace(account), GrantLevel::Read)
                .map_err(|err| Status::unavailable(format!("the identity table could not be read: {err}")))?;
            if inside {
                continue;
            }
            let public = ctx
                .control
                .namespace_for_outsider(&caller.principal_id, account)?
                .is_some_and(|ns| ns.events_public);
            if !public {
                return Err(not_shown(account));
            }
            outside.push(account.clone());
        }
        Ok(WatchScope::Accounts {
            names: namespaces.to_vec(),
            outside,
        })
    }

    /// Is `event` one this watcher may be handed? A server reader: yes. An
    /// account watcher: yes, unless it is about a repository in an account it
    /// is outside of and may not read.
    fn admits(&self, ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
        if let WatchScope::Mine = self {
            return Self::mine_admits(ctx, caller, event);
        }
        let WatchScope::Accounts { outside, .. } = self else {
            return Self::server_admits(ctx, caller, event);
        };
        match event.repo_path() {
            Some(path) if outside.contains(&path.namespace) => ctx
                .control
                .may(&caller.principal_id, &Access::repo(path), GrantLevel::Read)
                .unwrap_or(false),
            _ => true,
        }
    }
}

impl WatchScope {
    /// **A server reader's per-event boundary** (2026-10-05, the operator
    /// sees aggregates only): everything, except what happens inside a
    /// customer's account its server grant does not reach — that it is shown
    /// as any outsider is, only when the account admin made the events public.
    /// A lookup that fails is a `false`.
    fn server_admits(ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
        use gunnar_proto::event::Payload;
        match event.repo_path() {
            Some(path) => match ctx.shielded(caller, &path.namespace) {
                Ok(false) => true,
                Ok(true) => Self::mine_admits(ctx, caller, event),
                Err(_) => false,
            },
            None => match event.payload.as_ref() {
                Some(Payload::KeyUsed(used)) => match ctx.account_of_principal(&used.principal_id) {
                    Some(account) => !ctx.shielded(caller, &account).unwrap_or(true),
                    None => true,
                },
                _ => true,
            },
        }
    }

    /// [`WatchScope::Mine`]'s per-event boundary. A lookup that fails is a
    /// `false`: an event withheld is recoverable, one leaked is not.
    fn mine_admits(ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
        use gunnar_proto::event::Payload;
        match event.payload.as_ref() {
            Some(Payload::StreamOpened(_)) | Some(Payload::Gap(_)) => return true,
            _ => {}
        }
        let Some(path) = event.repo_path() else {
            return matches!(
                event.payload.as_ref(),
                Some(Payload::KeyUsed(used)) if used.principal_id == caller.principal_id
            );
        };
        let inside = ctx
            .control
            .may(&caller.principal_id, &Access::namespace(&path.namespace), GrantLevel::Read)
            .unwrap_or(false);
        if inside {
            return true;
        }
        let public = ctx
            .control
            .namespace_for_outsider(&caller.principal_id, &path.namespace)
            .ok()
            .flatten()
            .is_some_and(|ns| ns.events_public);
        public
            && ctx
                .control
                .may(&caller.principal_id, &Access::repo(path), GrantLevel::Read)
                .unwrap_or(false)
    }
}

impl Stream for Revalidated {
    type Item = Result<Event, Status>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        loop {
            if this.refused {
                return Poll::Ready(None);
            }
            // Poll first: a stream with nothing to hand over has nothing to check
            // an authorisation FOR, and re-checking a parked subscriber twice a
            // minute would be a control-plane read per idle console.
            let next = match Pin::new(&mut this.inner).poll_next(cx) {
                Poll::Ready(Some(item)) => item,
                other => return other,
            };
            if Instant::now() >= this.next_check {
                let rechecked = match &this.scope {
                    WatchScope::Server => this
                        .ctx
                        .require(&this.caller, Access::Server, GrantLevel::Read, "Events.Watch")
                        .map(|_| WatchScope::Server),
                    // The caller is still a principal here and not locked out
                    // — the question it was let in on. What each event may
                    // carry is decided per event, so a grant revoked
                    // mid-stream already stops the next delivery.
                    WatchScope::Mine => WatchScope::decide(&this.ctx, &this.caller, &[]).map(
                        |decided| match decided {
                            // A server grant arriving mid-stream does not
                            // widen an own watch either.
                            WatchScope::Server => WatchScope::Mine,
                            mine => mine,
                        },
                    ),
                    WatchScope::Accounts { names, .. } => {
                        WatchScope::decide(&this.ctx, &this.caller, names).map(|decided| {
                            match decided {
                                // A grant arriving mid-stream does not widen an
                                // account watcher: its boundary is what it asked
                                // for (and a server reader may read every
                                // repository, so nothing is filtered away).
                                WatchScope::Server => WatchScope::Accounts {
                                    names: names.clone(),
                                    outside: Vec::new(),
                                },
                                accounts => accounts,
                            }
                        })
                    }
                };
                match rechecked {
                    Ok(scope) => this.scope = scope,
                    Err(status) => {
                        tracing::info!(
                            principal = %this.caller.principal_id,
                            "closing an Events.Watch stream: its subscriber may no longer watch"
                        );
                        this.refused = true;
                        return Poll::Ready(Some(Err(status)));
                    }
                }
                this.next_check = Instant::now() + this.recheck;
            }
            match &next {
                Ok(event) if !this.scope.admits(&this.ctx, &this.caller, event) => continue,
                _ => return Poll::Ready(Some(next)),
            }
        }
    }
}

#[tonic::async_trait]
impl Events for EventsService {
    type WatchStream = Pin<Box<dyn Stream<Item = Result<Event, Status>> + Send + 'static>>;

    async fn watch(
        &self,
        request: Request<WatchRequest>,
    ) -> Result<Response<Self::WatchStream>, Status> {
        // **A server reader watches everything** (its `repos`/`namespaces` are
        // a FILTER). **Anybody else watches named accounts only**, and for them
        // the `namespaces` are the BOUNDARY: each must be an account the caller
        // is inside, or one whose account admin made its events public — and
        // even then a private repository's events stay inside its account
        // ([`WatchScope`]). An account's events are shown to nobody outside it
        // by default (owner, 2026-10-03).
        let caller = self.ctx.caller(&request)?;
        let scope = WatchScope::decide(&self.ctx, &caller, &request.get_ref().namespaces)?;
        let request = request.into_inner();
        if let WatchScope::Accounts { names, .. } = &scope {
            // A repository filter outside the named accounts would widen the
            // boundary through the other axis.
            if let Some(stray) = request.repos.iter().find(|r| !names.contains(&r.namespace)) {
                return Err(not_shown(&stray.namespace));
            }
        }

        // Validate every path in the filter rather than silently watching
        // nothing: a UI that mistyped a namespace should get an error, not an
        // eternally quiet stream it will read as "no activity".
        for repo in &request.repos {
            repo.validate()
                .map_err(|e| Status::invalid_argument(format!("`repos`: {e}")))?;
        }
        for kind in &request.kinds {
            gunnar_proto::EventKind::try_from(*kind)
                .map_err(|_| Status::invalid_argument(format!("unknown event kind {kind}")))?;
        }

        let filter = EventFilter::from_request(&request);
        tracing::debug!(
            namespaces = filter.namespaces.len(),
            repos = filter.repos.len(),
            kinds = filter.kinds.len(),
            from_seq = request.from_seq,
            "new Events.Watch subscriber"
        );

        // `subscribe_from`, not `subscribe`: every Watch stream opens with the
        // `StreamOpened` prologue, whether or not a cursor was sent. A client
        // that starts with no cursor is the one that most needs to be told
        // which epoch its first sequence number belongs to — it is going to
        // reconnect eventually, and by then the answer would be gone.
        // Wrapped, not returned bare: see [`Revalidated`]. `next_check` is
        // `now`, so even the first event out of a `from_seq` replay crosses a
        // fresh check rather than riding the one the handler just made — the
        // replay can be long, and "authorised when you asked" is precisely the
        // guarantee that was not enough.
        Ok(Response::new(Box::pin(Revalidated {
            inner: self.ctx.bus.subscribe_from(filter, request.from_seq),
            ctx: self.ctx.clone(),
            caller,
            scope,
            next_check: Instant::now(),
            recheck: RECHECK,
            refused: false,
        })))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::seam::MemoryControlPlane;
    use crate::{EventBus, ServerConfig};
    use gunnar_proto::{event, RepoDeleted, RepoPath};
    use std::sync::Arc;
    use tokio_stream::StreamExt as _;

    fn deleted(namespace: &str, repo: &str) -> event::Payload {
        event::Payload::RepoDeleted(RepoDeleted {
            repo: Some(RepoPath::new(namespace, repo)),
            principal_id: String::new(),
        })
    }

    /// A control plane holding one enrolled principal with a server-wide
    /// `Read` — the standing this RPC needs, and the standing the test takes
    /// away.
    fn plane_with_a_watcher() -> (Arc<MemoryControlPlane>, String) {
        use crate::seam::ControlPlane as _;
        let plane = MemoryControlPlane::new();
        // The principal has to be ENROLLED: `may` returns false for an id it
        // does not know, so granting without adding would make both tests below
        // pass for the wrong reason.
        let principal = plane
            .add_principal("watcher", "watcher", gunnar_proto::PrincipalKind::User)
            .expect("enrol the watcher");
        let plane = Arc::new(plane);
        plane.grant_server(&principal.id, GrantLevel::Read);
        (plane, principal.id)
    }

    fn caller(principal_id: &str) -> Caller {
        Caller {
            principal_id: principal_id.to_owned(),
            fingerprint: "SHA256:watcher".to_owned(),
            machine: false,
            authenticated: true,
        }
    }

    /// A stream that re-checks on every delivery, so a test can watch a
    /// revocation take effect without sleeping out [`RECHECK`].
    fn watching(ctx: &ServiceContext, principal_id: &str) -> Revalidated {
        Revalidated {
            inner: ctx.bus.subscribe_from(EventFilter::default(), 0),
            ctx: ctx.clone(),
            caller: caller(principal_id),
            scope: WatchScope::Server,
            next_check: Instant::now(),
            recheck: Duration::ZERO,
            refused: false,
        }
    }

    /// Pull items until one is an event carrying a payload, and return whether
    /// it arrived or the stream refused.
    ///
    /// The prologue (`StreamOpened`) is an event too, and skipping it by name
    /// rather than by counting keeps the assertions about the thing under test.
    async fn next_delivery(stream: &mut Revalidated) -> Result<Event, Status> {
        loop {
            match stream.next().await {
                Some(Ok(event)) if event.kind == gunnar_proto::EventKind::StreamOpened as i32 => {
                    continue
                }
                Some(item) => return item,
                None => panic!("the stream ended without delivering or refusing"),
            }
        }
    }

    /// **A watcher whose authorisation is withdrawn stops being handed
    /// events.**
    ///
    /// `Events.Watch` is the one RPC in the control plane whose answer outlives
    /// its request: `caller`/`require` ran when the pipe opened and the pipe
    /// then carried every event on the server for as long as the client kept
    /// reading. And it is the BROADEST read there is — the stream is
    /// server-scoped on purpose, so it carries pushes, deletions and repository
    /// names across every namespace, including ones the ex-subscriber never had
    /// access to.
    ///
    /// Both halves, because a stream that refused everything would satisfy the
    /// second on its own: an event published while the grant holds is
    /// DELIVERED, and one published after it is withdrawn is refused with a
    /// status rather than silently dropped — a pipe that has quietly stopped
    /// carrying reads as "no activity", which is the same failure the handler's
    /// filter validation refuses for a mistyped namespace.
    #[tokio::test]
    async fn a_watcher_who_may_no_longer_watch_stops_being_handed_events() {
        let (plane, watcher) = plane_with_a_watcher();
        let ctx = ServiceContext::new(
            plane.clone(),
            Arc::new(EventBus::new(64)),
            ServerConfig::default(),
        );
        let mut stream = watching(&ctx, &watcher);

        ctx.bus.publish(deleted("alpha", "one"));
        let event = next_delivery(&mut stream)
            .await
            .expect("a watcher who may watch is handed the event");
        assert_eq!(
            event.kind,
            gunnar_proto::EventKind::RepoDeleted as i32,
            "the fixture delivered something other than the published event"
        );

        // The grant goes. Nothing else changes — the stream is open, the bus is
        // the same bus, and the subscriber is still reading.
        plane.force_may(Some(false));
        ctx.bus.publish(deleted("alpha", "two"));

        let status = next_delivery(&mut stream)
            .await
            .expect_err("the stream kept delivering to a subscriber who may no longer watch");
        assert_eq!(
            status.code(),
            tonic::Code::PermissionDenied,
            "the stream ended, but not by refusing: {status:?}"
        );

        // Terminal: a refused stream must not be pollable for more. Without
        // this the refusal could be one skipped event rather than a closed
        // pipe, and the next event would flow.
        assert!(
            stream.next().await.is_none(),
            "a refused Events.Watch stream handed out something after its refusal"
        );
    }

    /// The identity-value guard for the test above.
    ///
    /// A `Revalidated` that refused on every delivery would satisfy the refusal
    /// half perfectly. This drives four events through a subscriber whose grant
    /// never changes and requires all four.
    #[tokio::test]
    async fn a_watcher_who_still_may_keeps_being_handed_events() {
        let (plane, watcher) = plane_with_a_watcher();
        let ctx = ServiceContext::new(
            plane.clone(),
            Arc::new(EventBus::new(64)),
            ServerConfig::default(),
        );
        let mut stream = watching(&ctx, &watcher);

        for n in 0..4 {
            ctx.bus.publish(deleted("alpha", &format!("repo-{n}")));
            next_delivery(&mut stream)
                .await
                .unwrap_or_else(|status| panic!("delivery {n} was refused: {status:?}"));
        }
    }
}