PiCloud/crates/manager-core/src/pubsub_repo.rs

//! `PubsubRepo` — publish-time fan-out for the v1.1.5 `pubsub::*` SDK.
//!
//! `publish_durable` writes one outbox row per matching enabled `pubsub`
//! trigger, all inside a single transaction so a partial fan-out (some
//! subscribers got rows, others didn't, then a crash) can't happen.
//! Each delivery row then retries / dead-letters independently through
//! the existing dispatcher — no pub/sub-specific dispatch branching.
//!
//! Topic pattern matching runs in Rust (`picloud_shared::topic_matches`)
//! against the small set of the app's enabled pubsub triggers, keeping
//! the SELECT trivial. v1.2 can add a topic-trie index if fan-out
//! becomes a hot path.

use async_trait::async_trait;
use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId, GroupId};
use sqlx::PgPool;
use uuid::Uuid;

use crate::config_resolver::CHAIN_LEVELS_CTE;

#[derive(Debug, thiserror::Error)]
pub enum PubsubRepoError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),
}

/// The execution-context bits a fan-out needs to stamp onto each outbox
/// row. Derived from the publishing script's `SdkCallCx`.
#[derive(Debug, Clone, Copy)]
pub struct PublishCtx {
    pub app_id: AppId,
    pub origin_principal: Option<AdminUserId>,
    pub trigger_depth: u32,
    pub root_execution_id: ExecutionId,
}

#[async_trait]
pub trait PubsubRepo: Send + Sync {
    /// Fan out a publish to every matching enabled pubsub trigger in
    /// `ctx.app_id`, inserting one outbox row each in a SINGLE
    /// transaction. `event_payload` is the serialized
    /// `TriggerEvent::Pubsub`. Returns the number of delivery rows
    /// written (0 when no trigger matched — the publish still succeeds).
    async fn fan_out_publish(
        &self,
        ctx: PublishCtx,
        topic: &str,
        event_payload: serde_json::Value,
    ) -> Result<u32, PubsubRepoError>;

    /// §11.6 D2: fan out a SHARED-topic publish to every `shared = true` pubsub
    /// trigger on `owning_group` (the group resolved from the shared-topic
    /// declaration), inserting one outbox row each stamped with the WRITER
    /// `ctx.app_id` — so the handler runs under the publishing app, matching the
    /// M2 shared-collection trigger model. No chain union (the owning group is
    /// already resolved) and no suppression (shared triggers aren't suppressible).
    async fn fan_out_shared_publish(
        &self,
        ctx: PublishCtx,
        owning_group: GroupId,
        topic: &str,
        event_payload: serde_json::Value,
    ) -> Result<u32, PubsubRepoError>;
}

pub struct PostgresPubsubRepo {
    pool: PgPool,
}

impl PostgresPubsubRepo {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

#[derive(sqlx::FromRow)]
struct PubsubTriggerRow {
    id: Uuid,
    script_id: Uuid,
    topic_pattern: String,
}

#[async_trait]
impl PubsubRepo for PostgresPubsubRepo {
    async fn fan_out_publish(
        &self,
        ctx: PublishCtx,
        topic: &str,
        event_payload: serde_json::Value,
    ) -> Result<u32, PubsubRepoError> {
        let mut tx = self.pool.begin().await?;

        // Load all enabled pubsub triggers for the app; filter by topic
        // pattern in Rust (keeps the query simple, honours the
        // empty/`*`/prefix semantics without teaching SQL about globs).
        // §11 tail: the chain union picks up the app's own pubsub triggers AND
        // ancestor-group pubsub TEMPLATES (live, no materialization). The outbox
        // row below stamps the firing `ctx.app_id`, so a template runs under the
        // publishing app — the inheriting-app boundary. The suppression
        // anti-join drops an inherited template whose handler the app opts out
        // of (`$1` = ctx.app_id).
        let rows: Vec<PubsubTriggerRow> = sqlx::query_as(&format!(
            "{CHAIN_LEVELS_CTE} \
             SELECT t.id, t.script_id, d.topic_pattern \
             FROM triggers t \
             JOIN pubsub_trigger_details d ON d.trigger_id = t.id \
             JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \
             WHERE t.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = FALSE{ANTIJOIN}",
            ANTIJOIN = crate::trigger_repo::TRIGGER_SUPPRESSION_ANTIJOIN,
        ))
        .bind(ctx.app_id.into_inner())
        .fetch_all(&mut *tx)
        .await?;

        let mut written: u32 = 0;
        for r in rows {
            if !topic_matches(&r.topic_pattern, topic) {
                continue;
            }
            insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?;
            written += 1;
        }

        // Commit once — all rows or none.
        tx.commit().await?;
        Ok(written)
    }

    async fn fan_out_shared_publish(
        &self,
        ctx: PublishCtx,
        owning_group: GroupId,
        topic: &str,
        event_payload: serde_json::Value,
    ) -> Result<u32, PubsubRepoError> {
        let mut tx = self.pool.begin().await?;

        // §11.6 D2: only `shared = true` pubsub triggers on the RESOLVED owning
        // group — the shared-topic namespace boundary (a per-app or non-owning
        // trigger never matches a shared publish). Topic-pattern match in Rust
        // like the per-app path.
        let rows: Vec<PubsubTriggerRow> = sqlx::query_as(
            "SELECT t.id, t.script_id, d.topic_pattern \
             FROM triggers t \
             JOIN pubsub_trigger_details d ON d.trigger_id = t.id \
             WHERE t.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = TRUE \
               AND t.group_id = $1",
        )
        .bind(owning_group.into_inner())
        .fetch_all(&mut *tx)
        .await?;

        let mut written: u32 = 0;
        for r in rows {
            if !topic_matches(&r.topic_pattern, topic) {
                continue;
            }
            insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?;
            written += 1;
        }

        tx.commit().await?;
        Ok(written)
    }
}

/// Insert one pubsub delivery row, stamped with the writer `ctx.app_id` so the
/// handler runs under the publishing app. Shared by the per-app + shared
/// fan-outs.
async fn insert_pubsub_outbox_row(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    ctx: &PublishCtx,
    trigger_id: Uuid,
    script_id: Uuid,
    event_payload: &serde_json::Value,
) -> Result<(), PubsubRepoError> {
    sqlx::query(
        "INSERT INTO outbox ( \
            app_id, source_kind, trigger_id, script_id, reply_to, \
            payload, origin_principal, trigger_depth, root_execution_id \
         ) VALUES ($1, 'pubsub', $2, $3, NULL, $4, $5, $6, $7)",
    )
    .bind(ctx.app_id.into_inner())
    .bind(trigger_id)
    .bind(script_id)
    .bind(event_payload)
    .bind(ctx.origin_principal.map(AdminUserId::into_inner))
    .bind(i32::try_from(ctx.trigger_depth.saturating_add(1)).unwrap_or(1))
    .bind(ctx.root_execution_id.into_inner())
    .execute(&mut **tx)
    .await?;
    Ok(())
}