//! `resolve_owning_group` — the isolation boundary for EVERY group shared
//! collection (kv, docs, files, topics, queues) — against real Postgres.
//!
//! ## Why this file exists
//!
//! `group_collection_repo`'s own doc calls the ancestor-chain walk "**the**
//! isolation boundary": a foreign app's chain never contains the owning group, so
//! the name does not resolve and the app gets `CollectionNotShared`.
//!
//! Every test that claimed to pin that boundary — `unrelated_app_gets_collection_
//! not_shared` in `group_kv_service` / `group_docs_service` / `group_files_service`,
//! and `realtime_authority`'s SSE 404 case — injects a **`FakeResolver` HashMap**
//! that the test itself populates. The foreign app resolves to `None` there only
//! because the test never inserted it. Those tests prove the services propagate a
//! `None`; they cannot see the SQL that produces it. So the actual boundary — the
//! recursive CTE — had no test at all: dropping `JOIN chain c ON gc.group_id =
//! c.group_owner` would let ANY app resolve ANY group's shared collection, a total
//! cross-tenant breach, and the whole unit suite would stay green.
//!
//! Every assertion below therefore drives the real query over a real tree:
//!
//! ```text
//!            root
//!         ┌───┴────┐
//!      shared     sibling          shared declares catalog(kv), articles(docs)
//!      ┌──┴──┐        │
//!    deep   app_mid  app_sibling   deep declares catalog(kv)  ← shadows
//!      │
//!   app_deep
//! ```
//!
//! Skips cleanly when `DATABASE_URL` is unset.

use picloud_manager_core::group_collection_repo::{insert_collection_tx, resolve_owning_group};
use picloud_shared::{AppId, GroupId, ScriptOwner};
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 {
        picloud_test_support::abort_if_db_required("group_collection_isolation");
        eprintln!("group_collection_isolation: DATABASE_URL unset — skipping");
        return None;
    };
    let pool = PgPoolOptions::new()
        .max_connections(4)
        .connect(&url)
        .await
        .expect("connect");
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("migrate");
    Some(pool)
}

async fn mk_group(pool: &PgPool, slug: &str, parent: Option<Uuid>) -> Uuid {
    let (id,): (Uuid,) = sqlx::query_as(
        "INSERT INTO groups (slug, name, parent_id) VALUES ($1, $1, $2) RETURNING id",
    )
    .bind(slug)
    .bind(parent)
    .fetch_one(pool)
    .await
    .expect("insert group");
    id
}

async fn mk_app(pool: &PgPool, slug: &str, group: Uuid) -> Uuid {
    let (id,): (Uuid,) =
        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
            .bind(slug)
            .bind(group)
            .fetch_one(pool)
            .await
            .expect("insert app");
    id
}

async fn declare(pool: &PgPool, owner: ScriptOwner, name: &str, kind: &str) {
    let mut tx = pool.begin().await.expect("begin");
    insert_collection_tx(&mut tx, owner, name, kind)
        .await
        .expect("declare collection");
    tx.commit().await.expect("commit");
}

struct Tree {
    root: Uuid,
    shared: Uuid,
    deep: Uuid,
    app_mid: Uuid,
    app_deep: Uuid,
    app_sibling: Uuid,
}

async fn build_tree(pool: &PgPool) -> Tree {
    let u = Uuid::new_v4().simple().to_string();
    let root = mk_group(pool, &format!("gci-root-{u}"), None).await;
    let shared = mk_group(pool, &format!("gci-shared-{u}"), Some(root)).await;
    let sibling = mk_group(pool, &format!("gci-sibling-{u}"), Some(root)).await;
    let deep = mk_group(pool, &format!("gci-deep-{u}"), Some(shared)).await;

    let app_mid = mk_app(pool, &format!("gci-app-mid-{u}"), shared).await;
    let app_deep = mk_app(pool, &format!("gci-app-deep-{u}"), deep).await;
    let app_sibling = mk_app(pool, &format!("gci-app-sib-{u}"), sibling).await;

    // The `shared` group shares two collections of DIFFERENT kinds under names
    // that let us prove the kind filter is real.
    declare(
        pool,
        ScriptOwner::Group(GroupId::from(shared)),
        "catalog",
        "kv",
    )
    .await;
    declare(
        pool,
        ScriptOwner::Group(GroupId::from(shared)),
        "articles",
        "docs",
    )
    .await;

    Tree {
        root,
        shared,
        deep,
        app_mid,
        app_deep,
        app_sibling,
    }
}

async fn cleanup(pool: &PgPool, t: &Tree) {
    // groups CASCADE to apps and to group_collections.
    let _ = sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(t.root)
        .execute(pool)
        .await;
}

async fn resolve(pool: &PgPool, app: Uuid, name: &str, kind: &str) -> Option<GroupId> {
    resolve_owning_group(pool, AppId::from(app), name, kind)
        .await
        .expect("resolve")
}

/// The whole point: a descendant resolves, a sibling-subtree app does not.
#[tokio::test]
async fn the_chain_walk_is_the_isolation_boundary() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let t = build_tree(&pool).await;

    let mid = resolve(&pool, t.app_mid, "catalog", "kv").await;
    let deep = resolve(&pool, t.app_deep, "catalog", "kv").await;
    let sibling = resolve(&pool, t.app_sibling, "catalog", "kv").await;
    cleanup(&pool, &t).await;

    // An app directly under the sharing group resolves it.
    assert_eq!(
        mid,
        Some(GroupId::from(t.shared)),
        "an app under the declaring group must resolve the shared collection"
    );
    // …and so does one further down the chain (the walk is recursive, not a
    // single parent hop).
    assert_eq!(
        deep,
        Some(GroupId::from(t.shared)),
        "a grandchild app must resolve an ancestor group's shared collection"
    );
    // THE BOUNDARY. `app_sibling` shares a common ancestor (root) with the
    // declaring group but is not beneath it. It must not resolve. Delete the
    // `JOIN chain c ON gc.group_id = c.group_owner` and this is the assertion
    // that fires.
    assert_eq!(
        sibling, None,
        "a sibling-subtree app must NOT resolve another subtree's shared collection"
    );
}

/// A `kv` and a `docs` collection of the same name are distinct stores, so the
/// `kind` filter is part of the boundary, not a convenience.
#[tokio::test]
async fn the_kind_filter_is_part_of_the_boundary() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let t = build_tree(&pool).await;

    let docs_ok = resolve(&pool, t.app_mid, "articles", "docs").await;
    let docs_as_kv = resolve(&pool, t.app_mid, "articles", "kv").await;
    let kv_as_docs = resolve(&pool, t.app_mid, "catalog", "docs").await;
    let undeclared = resolve(&pool, t.app_mid, "nonexistent", "kv").await;
    cleanup(&pool, &t).await;

    assert_eq!(
        docs_ok,
        Some(GroupId::from(t.shared)),
        "the docs collection resolves under its own kind"
    );
    assert_eq!(
        docs_as_kv, None,
        "a docs collection must NOT resolve as kv — the kinds are distinct stores"
    );
    assert_eq!(kv_as_docs, None, "a kv collection must NOT resolve as docs");
    assert_eq!(undeclared, None, "an undeclared name resolves to nothing");
}

/// Nearest-wins. The doc calls this "security-relevant": a nearer group's
/// declaration shadows a farther one, so the data a script reaches is the nearer
/// store. If `ORDER BY depth ASC LIMIT 1` were dropped (or reversed), a deep app
/// would silently read/write the WRONG group's shared store.
#[tokio::test]
async fn a_nearer_declaration_shadows_a_farther_one() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let t = build_tree(&pool).await;

    // `deep` (between `shared` and `app_deep`) now declares the same name+kind.
    declare(
        &pool,
        ScriptOwner::Group(GroupId::from(t.deep)),
        "catalog",
        "kv",
    )
    .await;

    let deep = resolve(&pool, t.app_deep, "catalog", "kv").await;
    let mid = resolve(&pool, t.app_mid, "catalog", "kv").await;
    cleanup(&pool, &t).await;

    assert_eq!(
        deep,
        Some(GroupId::from(t.deep)),
        "the NEAREST declaring ancestor wins — app_deep must reach deep's store, not shared's"
    );
    // The shadowing is scoped to the shadowed subtree: app_mid is not under
    // `deep`, so it still resolves to `shared`.
    assert_eq!(
        mid,
        Some(GroupId::from(t.shared)),
        "an app outside the shadowing group's subtree still resolves the farther one"
    );
}

/// "The join is on `group_owner` only: an app-declared marker never makes a
/// collection visible to *other* apps — sharing is a group property." Pin that:
/// an app-owned marker must resolve for NOBODY, including its own declaring app.
#[tokio::test]
async fn an_app_owned_marker_shares_with_nobody() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let t = build_tree(&pool).await;

    declare(
        &pool,
        ScriptOwner::App(AppId::from(t.app_mid)),
        "private",
        "kv",
    )
    .await;

    let own = resolve(&pool, t.app_mid, "private", "kv").await;
    let other = resolve(&pool, t.app_deep, "private", "kv").await;
    cleanup(&pool, &t).await;

    assert_eq!(
        own, None,
        "an app-declared marker is not a SHARED collection — not even for the declaring app"
    );
    assert_eq!(other, None, "and certainly not for anyone else");
}

/// The query compares `LOWER(gc.name) = LOWER($2)`, so lookup is
/// case-insensitive. Worth pinning: if it silently became case-SENSITIVE, a
/// script using a differently-cased name would get `CollectionNotShared` and the
/// failure would look like a permissions bug.
#[tokio::test]
async fn resolution_is_case_insensitive() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let t = build_tree(&pool).await;

    let mixed = resolve(&pool, t.app_mid, "CaTaLoG", "kv").await;
    cleanup(&pool, &t).await;

    assert_eq!(
        mixed,
        Some(GroupId::from(t.shared)),
        "the name match is case-insensitive"
    );
}