skilj-core 0.2.0

//! Cross-instance push completeness (issue #2; `@guarantee
//! DeliverySpansInstances`/`RegistrationReachesEveryInstance` in
//! specs/skilj.allium) - see docs/architecture.md's own write-up of this
//! pass for the full design and its narrower spec `Excludes` entry for
//! what stays deliberately out of scope.
//!
//! Three Postgres `NOTIFY` channels: this crate's own `db` module sends
//! on them (`db::notify_event_appended`/`db::notify_registration_changed`/
//! `db::notify_revocation`, called at the same choke points
//! `EventBroadcaster::publish`/`RevocationBroadcaster::publish`/every
//! type-registration write already go through), and this module's
//! [`Listener`] receives on them. Every instance both sends and
//! receives, symmetrically - coordinated purely through the shared
//! Postgres database `PgListener` already connects to, never through
//! instance discovery, a leader, or any other shared infrastructure.
//! `skilj` (the facade crate) is what actually drives a [`Listener`] in
//! a background task and dispatches each [`Message`] into the right
//! destination (`EventBroadcaster`/`RevocationBroadcaster`/a GraphQL
//! schema rebuild) - this module only owns the `LISTEN` mechanics and
//! the channel-name/payload contract, staying inside "skilj-core is the
//! only crate that owns the database driver" (docs/architecture.md
//! §3.1); it has no reason to know `skilj-graphql`'s `SchemaRegistry`
//! exists.
//!
//! **Delivery is at-most-once, deliberately** - any notification sent
//! while the listening connection is down is gone for good. Every
//! message here is a liveliness signal only, never a correctness-bearing
//! write; Postgres remains the durable source of truth. What makes that
//! safe is that a loss is *noticed*: [`Listener::recv`] reconnects
//! explicitly and returns [`Message::Resync`], and the caller re-reads
//! everything the channels drive (docs/architecture.md §83). It used to
//! rely on `PgListener::recv`'s transparent reconnect, which hid the
//! loss entirely: a registration change made meanwhile left this
//! instance's schema stale until the next one arrived, and local event
//! subscribers silently skipped the missed events.

use crate::access_control::RevokedMapping;
use crate::db::Pool;

/// `db::notify_event_appended`'s channel.
pub const EVENTS_CHANNEL: &str = "skilj_events";
/// `db::notify_revocation`'s channel.
pub const REVOCATIONS_CHANNEL: &str = "skilj_revocations";
/// `db::notify_registration_changed`'s channel.
pub const REGISTRATION_CHANGED_CHANNEL: &str = "skilj_registration_changed";

/// One decoded cross-instance notification - see this module's own doc
/// comment for the three channels these come from. Deliberately not
/// `Event`/`RevokedMapping` themselves reused verbatim as the payload
/// shape: `EventAppended` is a pointer (see its own doc comment for
/// why), and `RegistrationChanged` carries nothing at all.
///
/// `EventAppended` and `Revoked` both carry `origin_instance_id` - the
/// sending instance's own `EventBroadcaster::instance_id`/
/// `RevocationBroadcaster::instance_id` at the moment it published
/// locally and NOTIFYed, alongside each other, from the same choke
/// point. This module stays a decode-only layer (see its own doc
/// comment - it "has no reason to know skilj-graphql's SchemaRegistry
/// exists"), so it makes no decision based on this field itself; it's
/// the caller's own local broadcasters this needs comparing against, and
/// `skilj::SkiljBuilder::build`'s own dispatch loop is where that
/// comparison happens - see its own doc comment for why a match there
/// means "skip, this is my own write echoing back" rather than
/// "republish." `RegistrationChanged` carries no such field: unlike the
/// other two channels, nothing publishes a schema rebuild directly on
/// the write path - this NOTIFY is the *only* way any instance, including
/// the one that made the change, ever rebuilds its schema, so a
/// same-instance echo of it must always be acted on, never skipped.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum Message {
    /// `skilj_events` - a pointer, not the full event (Postgres's own
    /// 8000-byte `NOTIFY` payload cap, and every subscriber already
    /// re-fetches/renders the real event via the existing DB-backed path
    /// anyway). The receiving instance fetches the real event with this
    /// (via the plain, uncached `db::get_event_by_sequence` -
    /// `skilj::SkiljBuilder::build`'s own dispatch loop explains why not
    /// the cached variant) and republishes it into its own local
    /// `EventBroadcaster` - unless `origin_instance_id` names this exact
    /// instance, see this enum's own doc comment.
    EventAppended {
        bounded_context: String,
        sequence: i64,
        origin_instance_id: String,
    },
    /// `skilj_revocations` - small enough to carry in full.
    Revoked {
        revoked: RevokedMapping,
        origin_instance_id: String,
    },
    /// `skilj_registration_changed` - a bare signal, no payload: which
    /// exact type or bounded context changed doesn't matter, every
    /// listener reacts identically (rebuild the whole schema).
    RegistrationChanged,
    /// Not a channel: the listening connection dropped and has been
    /// re-established (and re-`LISTEN`ed) since the previous message, so
    /// any notification sent in between was lost. Everything the three
    /// channels drive has to be treated as possibly stale - rebuild the
    /// schema, and tell local subscribers they may have missed events
    /// (docs/architecture.md §83).
    Resync,
}

#[derive(serde::Deserialize)]
struct EventAppendedPayload {
    bounded_context: String,
    sequence: i64,
    /// `#[serde(default)]` so an older instance's payload (predating
    /// this field) still parses instead of being dropped as malformed -
    /// this module's own established "a gap self-heals, isn't fatal"
    /// principle. Defaulting to `""` (never a real
    /// `EventBroadcaster::instance_id`, always a fresh random id) means
    /// such a message is simply never mistaken for a self-echo, which
    /// only briefly reintroduces the pre-fix double-delivery this field
    /// exists to prevent, self-healing away as soon as every instance in
    /// a rolling deploy is running code that sends it.
    #[serde(default)]
    origin_instance_id: String,
}

#[derive(serde::Deserialize)]
struct RevokedPayload {
    role_id: String,
    bounded_context: String,
    /// See `EventAppendedPayload::origin_instance_id`'s own doc comment.
    #[serde(default)]
    origin_instance_id: String,
}

/// Thin wrapper around `sqlx::postgres::PgListener`, subscribed to all
/// three channels from [`connect`](Listener::connect) onward - callers
/// only ever see already-decoded [`Message`]s, never raw channel names
/// or JSON payloads.
pub struct Listener {
    pool: Pool,
    /// `None` between a detected connection loss and the reconnect.
    inner: Option<sqlx::postgres::PgListener>,
}

impl Listener {
    /// Opens one dedicated listening connection (from `pool`) and
    /// subscribes to all three channels. One `Listener` per `Skilj`
    /// instance is the intended shape - see `skilj::SkiljBuilder::build`'s
    /// own call site.
    pub async fn connect(pool: &Pool) -> crate::error::Result<Self> {
        Ok(Self {
            pool: pool.clone(),
            inner: Some(Self::listen(pool).await?),
        })
    }

    async fn listen(pool: &Pool) -> crate::error::Result<sqlx::postgres::PgListener> {
        let mut listener = sqlx::postgres::PgListener::connect_with(pool).await?;
        listener
            .listen_all([
                EVENTS_CHANNEL,
                REVOCATIONS_CHANNEL,
                REGISTRATION_CHANGED_CHANNEL,
            ])
            .await?;
        Ok(listener)
    }

    /// Blocks until the next cross-instance message. `PgListener::recv`
    /// already auto-reconnects and re-subscribes to every channel on a
    /// dropped connection (its own doc comment) - callers get that for
    /// free just by looping on this. A notification this process can't
    /// parse (an unknown channel, or a payload from some future/older
    /// version's shape) is logged and skipped, never a panic and never
    /// something that stops the loop over one bad message - the same
    /// "a gap self-heals, isn't fatal" principle every channel here
    /// already follows.
    ///
    /// A dropped connection is *not* reconnected transparently (as
    /// `PgListener::recv` would): `try_recv` reports it, this reconnects
    /// and re-`LISTEN`s first, and only then returns [`Message::Resync`] -
    /// so whatever the caller re-reads in response can't miss a change
    /// made after it. If reconnecting fails, the error is returned and
    /// the next call tries again.
    pub async fn recv(&mut self) -> crate::error::Result<Message> {
        loop {
            let listener = match &mut self.inner {
                Some(listener) => listener,
                None => {
                    self.inner = Some(Self::listen(&self.pool).await?);
                    return Ok(Message::Resync);
                }
            };
            let received = match listener.try_recv().await {
                Ok(received) => received,
                Err(err) => {
                    // Whatever went wrong, the connection can't be trusted
                    // not to have missed something: the next call
                    // reconnects and reports `Resync`.
                    self.inner = None;
                    return Err(err.into());
                }
            };
            let Some(notification) = received else {
                tracing::warn!("cross-instance listener connection lost, reconnecting");
                self.inner = None;
                continue;
            };
            match notification.channel() {
                EVENTS_CHANNEL => {
                    match serde_json::from_str::<EventAppendedPayload>(notification.payload()) {
                        Ok(p) => {
                            return Ok(Message::EventAppended {
                                bounded_context: p.bounded_context,
                                sequence: p.sequence,
                                origin_instance_id: p.origin_instance_id,
                            })
                        }
                        Err(err) => tracing::warn!(
                            error = %err,
                            "malformed skilj_events NOTIFY payload, skipping"
                        ),
                    }
                }
                REVOCATIONS_CHANNEL => {
                    match serde_json::from_str::<RevokedPayload>(notification.payload()) {
                        Ok(p) => {
                            return Ok(Message::Revoked {
                                revoked: RevokedMapping {
                                    role_id: p.role_id,
                                    bounded_context: p.bounded_context,
                                },
                                origin_instance_id: p.origin_instance_id,
                            })
                        }
                        Err(err) => tracing::warn!(
                            error = %err,
                            "malformed skilj_revocations NOTIFY payload, skipping"
                        ),
                    }
                }
                REGISTRATION_CHANGED_CHANNEL => return Ok(Message::RegistrationChanged),
                other => {
                    tracing::warn!(channel = other, "unexpected NOTIFY channel, skipping")
                }
            }
        }
    }
}