PiCloud/crates/manager-core/tests/shared_topics.rs

//! §11.6 D2 integration test: SHARED-topic pubsub fan-out.
//!
//! A group-owned `shared = true` pubsub trigger watches the group's shared
//! topic. A SHARED publish (`fan_out_shared_publish` on the owning group) fans
//! out to it — but NOT to a per-app pubsub trigger or a non-shared group
//! template of the same pattern. Conversely a per-app publish
//! (`fan_out_publish`) fans out to the app's own + inherited non-shared group
//! templates but NEVER the shared trigger (`shared` is the namespace boundary).
//!
//! Deterministic: drives the repo fan-out directly (no async dispatcher). Skips
//! when `DATABASE_URL` is unset.

#![allow(clippy::too_many_lines)]

use picloud_manager_core::pubsub_repo::{PostgresPubsubRepo, PublishCtx, PubsubRepo};
use picloud_shared::{AppId, ExecutionId, GroupId};
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use uuid::Uuid;

async fn pool_or_skip() -> Option<PgPool> {
    let Ok(url) = std::env::var("DATABASE_URL") else {
        eprintln!("shared_topics: DATABASE_URL unset — skipping");
        return None;
    };
    let pool = PgPoolOptions::new()
        .max_connections(2)
        .connect(&url)
        .await
        .expect("connect");
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("migrate");
    Some(pool)
}

/// Insert a pubsub trigger (owner + `shared` flag) + its details; returns id.
async fn pubsub_trigger(
    pool: &PgPool,
    app_id: Option<Uuid>,
    group_id: Option<Uuid>,
    script: Uuid,
    admin: Uuid,
    shared: bool,
    pattern: &str,
) -> Uuid {
    let row: (Uuid,) = sqlx::query_as(
        "INSERT INTO triggers \
            (app_id, group_id, script_id, kind, enabled, dispatch_mode, \
             retry_max_attempts, retry_backoff, retry_base_ms, \
             registered_by_principal, name, shared) \
         VALUES ($1, $2, $3, 'pubsub', TRUE, 'async', 3, 'exponential', 1000, $4, $5, $6) \
         RETURNING id",
    )
    .bind(app_id)
    .bind(group_id)
    .bind(script)
    .bind(admin)
    .bind(Uuid::new_v4().simple().to_string())
    .bind(shared)
    .fetch_one(pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)")
        .bind(row.0)
        .bind(pattern)
        .execute(pool)
        .await
        .unwrap();
    row.0
}

async fn outbox_count(pool: &PgPool, trigger_id: Uuid) -> i64 {
    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM outbox WHERE trigger_id = $1")
        .bind(trigger_id)
        .fetch_one(pool)
        .await
        .unwrap();
    n
}

fn ctx(app: Uuid) -> PublishCtx {
    PublishCtx {
        app_id: AppId::from(app),
        origin_principal: None,
        trigger_depth: 0,
        root_execution_id: ExecutionId::new(),
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shared_topic_fan_out_respects_the_namespace_boundary() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let sfx = Uuid::new_v4().simple().to_string();
    let admin: (Uuid,) = sqlx::query_as(
        "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
    )
    .bind(format!("st-{sfx}"))
    .fetch_one(&pool)
    .await
    .unwrap();
    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
        .bind(format!("st-g-{sfx}"))
        .fetch_one(&pool)
        .await
        .unwrap();
    let handler: (Uuid,) = sqlx::query_as(
        "INSERT INTO scripts (name, source, group_id) VALUES ($1, 'x', $2) RETURNING id",
    )
    .bind(format!("onmsg-{sfx}"))
    .bind(g.0)
    .fetch_one(&pool)
    .await
    .unwrap();
    let app: (Uuid,) =
        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
            .bind(format!("st-a-{sfx}"))
            .bind(g.0)
            .fetch_one(&pool)
            .await
            .unwrap();
    let app_script: (Uuid,) = sqlx::query_as(
        "INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2) RETURNING id",
    )
    .bind(format!("appmsg-{sfx}"))
    .bind(app.0)
    .fetch_one(&pool)
    .await
    .unwrap();

    // Three triggers, all watching `events.*`:
    let shared = pubsub_trigger(&pool, None, Some(g.0), handler.0, admin.0, true, "events.*").await;
    let group_tmpl = pubsub_trigger(
        &pool,
        None,
        Some(g.0),
        handler.0,
        admin.0,
        false,
        "events.*",
    )
    .await;
    let per_app = pubsub_trigger(
        &pool,
        Some(app.0),
        None,
        app_script.0,
        admin.0,
        false,
        "events.*",
    )
    .await;

    let repo = PostgresPubsubRepo::new(pool.clone());
    let payload = serde_json::json!({"topic": "events.created", "message": 1});

    // SHARED publish → only the shared trigger fires.
    let n = repo
        .fan_out_shared_publish(
            ctx(app.0),
            GroupId::from(g.0),
            "events.created",
            payload.clone(),
        )
        .await
        .unwrap();
    assert_eq!(n, 1, "shared publish matched exactly the shared trigger");
    assert_eq!(outbox_count(&pool, shared).await, 1, "shared trigger fired");
    assert_eq!(
        outbox_count(&pool, group_tmpl).await,
        0,
        "a non-shared group template must NOT fire on a shared publish"
    );
    assert_eq!(
        outbox_count(&pool, per_app).await,
        0,
        "a per-app trigger must NOT fire on a shared publish"
    );

    // PER-APP publish → the app's own + inherited non-shared template fire, but
    // NEVER the shared trigger.
    repo.fan_out_publish(ctx(app.0), "events.created", payload)
        .await
        .unwrap();
    assert_eq!(
        outbox_count(&pool, shared).await,
        1,
        "the shared trigger must NOT fire on a per-app publish (still 1 from before)"
    );
    assert_eq!(
        outbox_count(&pool, group_tmpl).await,
        1,
        "the inherited non-shared group template fires on a per-app publish"
    );
    assert_eq!(
        outbox_count(&pool, per_app).await,
        1,
        "the per-app trigger fires on a per-app publish"
    );

    // Cleanup.
    let _ = sqlx::query("DELETE FROM triggers WHERE registered_by_principal = $1")
        .bind(admin.0)
        .execute(&pool)
        .await;
    let _ = sqlx::query("DELETE FROM apps WHERE group_id = $1")
        .bind(g.0)
        .execute(&pool)
        .await;
    let _ = sqlx::query("DELETE FROM scripts WHERE group_id = $1")
        .bind(g.0)
        .execute(&pool)
        .await;
    let _ = sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(g.0)
        .execute(&pool)
        .await;
    let _ = sqlx::query("DELETE FROM admin_users WHERE id = $1")
        .bind(admin.0)
        .execute(&pool)
        .await;
}