PiCloud/crates/shared/src/pubsub.rs

//! `PubsubService` — the v1.1.5 durable pub/sub contract.
//!
//! `pubsub::publish_durable(topic, message)` writes to the universal
//! outbox; the publish-time fan-out inserts one delivery row per
//! matching `pubsub` trigger, and each delivery retries / dead-letters
//! independently (the dispatcher already handles one-row-equals-one-
//! dispatch — no dispatcher changes for pub/sub).
//!
//! `publish_ephemeral` is committed as a v1.2 addition — the suffix
//! naming exists now so users learn "durable by default" from day one.
//!
//! Topic pattern matching runs in Rust (not SQL) so the trigger-select
//! query stays simple. The matcher + validator live here in
//! `picloud-shared` so the manager-core publish path, the admin trigger
//! endpoint, and tests all agree on the rules.

use async_trait::async_trait;
use thiserror::Error;

use crate::SdkCallCx;

#[async_trait]
pub trait PubsubService: Send + Sync {
    /// Durable publish: writes the message to the outbox, fanned out to
    /// every matching enabled `pubsub` trigger in `cx.app_id`. Succeeds
    /// silently (zero rows written) when no trigger matches the topic.
    async fn publish_durable(
        &self,
        cx: &SdkCallCx,
        topic: &str,
        message: serde_json::Value,
    ) -> Result<(), PubsubError>;

    /// Mint an HMAC-signed realtime subscriber token (v1.1.6). Backs the
    /// `pubsub::subscriber_token(topics, ttl)` Rhai SDK call. The minted
    /// token authorizes an external SSE client to subscribe to the given
    /// `topics` for `ttl_seconds` (clamped to the configured bounds; the
    /// configured default applies when `ttl_seconds` is `None`).
    ///
    /// Every topic must already be registered as externally subscribable
    /// in `cx.app_id`; `cx.principal` must be `Some` (anonymous
    /// public-HTTP scripts can't mint). See [`PubsubError::SubscriberToken`]
    /// for the rejection messages.
    ///
    /// The default impl errors `Unavailable` so test fakes and the
    /// `NoopPubsubService` keep compiling; the real minting lives in
    /// manager-core's `PubsubServiceImpl`.
    async fn mint_subscriber_token(
        &self,
        cx: &SdkCallCx,
        topics: Vec<String>,
        ttl_seconds: Option<i64>,
    ) -> Result<String, PubsubError> {
        let _ = (cx, topics, ttl_seconds);
        Err(PubsubError::Backend(
            "subscriber tokens are not wired in".into(),
        ))
    }
}

#[derive(Debug, Error)]
pub enum PubsubError {
    /// Empty topic; rejected at the SDK boundary.
    #[error("topic must not be empty")]
    EmptyTopic,

    /// Caller principal lacked the required capability. Only raised when
    /// `cx.principal.is_some()` (script-as-gate; public HTTP skips it).
    #[error("forbidden")]
    Forbidden,

    /// Serialization / validation failure on the message.
    #[error("pubsub rejected: {0}")]
    Rejected(String),

    /// JSON-encoded message exceeded the configured `max_message_bytes`
    /// cap (`PICLOUD_PUBSUB_MAX_MESSAGE_BYTES`). Defends against an
    /// anonymous public script amplifying one publish into N outbox rows
    /// × payload bytes.
    #[error("pubsub: message too large ({actual} bytes; limit {limit})")]
    MessageTooLarge { limit: usize, actual: usize },

    /// A `pubsub::subscriber_token` mint was rejected (empty topics,
    /// unregistered topic, ttl out of range, anonymous caller). The
    /// string is the full user-facing message; the SDK surfaces it
    /// verbatim so scripts see the documented wording.
    #[error("{0}")]
    SubscriberToken(String),

    /// Anything else — Postgres unavailable, etc. Named `Backend` to
    /// align with KvError / DocsError / FilesError / SecretsError.
    #[error("pubsub backend error: {0}")]
    Backend(String),
}

/// Match a stored `topic_pattern` against a published `topic`.
///
/// - `"*"` matches every topic.
/// - `"<prefix>.*"` matches any topic starting with `"<prefix>."`.
/// - anything else is an exact match.
///
/// Mid-pattern wildcards (`*.created`, `a.*.b`) are NOT supported — they
/// are rejected at trigger creation by [`validate_topic_pattern`], so
/// the only patterns reaching this matcher are exact / prefix / `*`.
#[must_use]
pub fn topic_matches(pattern: &str, topic: &str) -> bool {
    if pattern == "*" {
        return true;
    }
    if let Some(prefix) = pattern.strip_suffix('*') {
        // `prefix` retains the trailing '.', e.g. "user." for "user.*".
        return topic.starts_with(prefix);
    }
    pattern == topic
}

/// Validate a subscription topic pattern. Accepts exactly: `"*"`
/// (universal), `"<prefix>.*"` (prefix wildcard, single trailing star),
/// or a literal with no `*` (exact). Everything else — mid-pattern
/// wildcards, multiple stars, a star not at the end — is rejected.
///
/// # Errors
///
/// Returns `Err(message)` with `"unsupported pubsub topic pattern: …"`
/// for any unsupported shape (or an empty pattern).
pub fn validate_topic_pattern(pattern: &str) -> Result<(), String> {
    if pattern.is_empty() {
        return Err("unsupported pubsub topic pattern: <empty>".to_string());
    }
    if pattern == "*" {
        return Ok(());
    }
    let stars = pattern.matches('*').count();
    if stars == 0 {
        return Ok(()); // exact
    }
    if stars == 1 && pattern.ends_with(".*") {
        return Ok(()); // prefix wildcard
    }
    Err(format!("unsupported pubsub topic pattern: {pattern}"))
}

/// Stub for the test harness so executor-core integration tests can
/// build a `Services` bundle without a database. Every call errors.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopPubsubService;

#[async_trait]
impl PubsubService for NoopPubsubService {
    async fn publish_durable(
        &self,
        _cx: &SdkCallCx,
        _topic: &str,
        _message: serde_json::Value,
    ) -> Result<(), PubsubError> {
        Err(PubsubError::Backend("pubsub is not wired in".into()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn exact_match() {
        assert!(topic_matches("user.created", "user.created"));
        assert!(!topic_matches("user.created", "user.deleted"));
        assert!(!topic_matches("user.created", "user.created.x"));
    }

    #[test]
    fn prefix_wildcard() {
        assert!(topic_matches("user.*", "user.created"));
        assert!(topic_matches("user.*", "user.deleted"));
        assert!(!topic_matches("user.*", "users.created"));
        assert!(!topic_matches("user.*", "order.created"));
    }

    #[test]
    fn universal() {
        assert!(topic_matches("*", "anything"));
        assert!(topic_matches("*", "a.b.c"));
    }

    #[test]
    fn validation_accepts_supported_shapes() {
        assert!(validate_topic_pattern("*").is_ok());
        assert!(validate_topic_pattern("user.created").is_ok());
        assert!(validate_topic_pattern("user.*").is_ok());
        assert!(validate_topic_pattern("a.b.c").is_ok());
    }

    #[test]
    fn validation_rejects_unsupported_shapes() {
        for bad in ["*.created", "**", "a.*.b", "user.*x", "*user", ""] {
            assert!(
                validate_topic_pattern(bad).is_err(),
                "expected {bad:?} to be rejected"
            );
        }
    }
}