nordisk/monetize

crates/monetize-server/src/services/events.rs

//! `Events.Watch`, the server-streaming feed.
//!
//! The whole implementation is "build a filter, subscribe to the bus, box the
//! stream". Everything interesting — the bounded ring, 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.

use std::pin::Pin;
use std::sync::Arc;

use monetize_proto::events_server::Events;
use monetize_proto::{self as pb};
use tokio_stream::Stream;
use tonic::{Request, Response, Status};

use crate::bus::EventFilter;
use crate::state::AppState;

pub struct EventsService(pub Arc<AppState>);

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

    async fn watch(
        &self,
        request: Request<pb::WatchRequest>,
    ) -> Result<Response<Self::WatchStream>, Status> {
        // Server-scoped on purpose, exactly as gunnar's is. The bus carries
        // every product, and `products` / `tenants` are a FILTER, not a
        // boundary: a subscriber that narrowed to one tenant is still sitting on
        // a stream it could widen by sending a second request. The bearer that
        // reaches this method is the suite bearer, which already reads every
        // tenant over `Entitlement.List`; until the bus can be narrowed per
        // principal, watching is a server-wide read and is documented as one.
        let request = request.into_inner();

        // Refuse an unknown kind rather than silently watching nothing: a
        // console built against a newer proto should get an error, not an
        // eternally quiet stream it will render as "no activity".
        for kind in &request.kinds {
            pb::EventKind::try_from(*kind)
                .map_err(|_| Status::invalid_argument(format!("unknown event kind {kind}")))?;
        }

        let filter = EventFilter::from_request(&request);
        eprintln!(
            "monetize-server: Events.Watch subscriber products={} tenants={} kinds={} from_seq={}",
            filter.products.len(),
            filter.tenants.len(),
            filter.kinds.len(),
            request.from_seq
        );

        // `subscribe_from`, not `subscribe`: every Watch stream opens with the
        // 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 will reconnect eventually, and by then
        // the answer would be gone.
        Ok(Response::new(Box::pin(
            self.0.bus.subscribe_from(filter, request.from_seq),
        )))
    }
}