axond 0.3.39

//! The fixtures the isolation scenarios share: a real journal, sessions pinned
//! to one tenant, and the two-tenant state every scenario is stated against.
//!
//! # A schema per scenario, dropped with the fixture
//!
//! Every scenario owns a PostgreSQL schema, migrates the real journal into it,
//! and drops it — and the login roles it created — in [`Drop`], so a failing
//! assertion leaves nothing behind for the next run to inherit. Roles are
//! cluster-wide rather than schema-scoped, which is precisely why they are
//! tracked and dropped here: a leaked `LOGIN` role from a panicking test is a
//! credential on the CI database that outlives the test that made it.
//!
//! # Why the pinned session is a separate role
//!
//! A superuser bypasses row-level security unconditionally, and the schema's
//! owner is the connection the tests migrate with. So a scenario that asserted
//! about RLS through the store's own connection would assert nothing at all:
//! [`Journal::session`] therefore creates an ordinary `LOGIN` role, grants it
//! exactly the privileges the scenario needs, connects as it, and pins it with
//! `SET axond.tenant_id`. That is the shape a deployment gets when its
//! application role is not the migrating role — the shape the policies in
//! `control_plane_0002_tenancy_access.sql` were written for.
//!
//! # Absence is asserted by exact name, not by fragment
//!
//! [`Absent`] looks for exact identifiers rather than reusing
//! [`LeakSweep`](crate::secret_redaction::sweep::LeakSweep). A leak sweep also
//! matches twelve-character fragments, which is right for high-entropy key
//! material and wrong here: the fixtures' ids are derived from small seeds, so
//! two tenants' ids differ in a few characters and share every fragment. A
//! fragment match would report a tenant's own id as its neighbour's. What these
//! scenarios need is narrower and exact — *this* tenant id, project id, slug,
//! principal id or credential id does not appear in what the caller was told.

use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use ring::rand::{SecureRandom, SystemRandom};
use tokio_postgres::{Client, Config};

use crate::backends::control_plane::postgres::{ControlPlaneSettings, PostgresControlPlane};
use crate::backends::control_plane::{ControlPlaneError, ControlPlaneStore};
use crate::desired_state::{
    DesiredState, ExpectedRevision, LoadedRevision, RevisionId, TenantId, fixtures,
};

/// A password for one fixture's roles, from the system generator.
///
/// Nothing in a scenario needs the password to be knowable, and a literal in
/// the source would be: [`Drop`] is the only thing that removes these roles, so
/// a killed test process (CI cancellation, OOM) leaves a `LOGIN` role behind on
/// a shared database. Left behind with a password nobody has, it is unusable
/// rather than an open door.
fn role_password() -> String {
    let mut bytes = [0u8; 24];
    SystemRandom::new()
        .fill(&mut bytes)
        .expect("the system random generator");
    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}

/// The tenant every scenario calls as: `acme`, seed 1 of the domain fixtures.
pub(crate) fn caller() -> TenantId {
    fixtures::tenant_id(1)
}

/// The tenant every scenario must fail to reach: `globex`, seed 11.
pub(crate) fn other() -> TenantId {
    fixtures::tenant_id(11)
}

/// Two tenants, each with a project, a directory, a credential, an alias and a
/// policy of its own, and nothing of the other's.
///
/// Built on [`fixtures::two_tenant_directory_state`] and extended with the
/// surfaces #225 names — credentials, aliases, policies — because a projection
/// isolation claim about tenancy alone would leave exactly the rows an operator
/// worries about untested. Neither tenant references anything of the other's, so
/// a cross-tenant edge in any scenario below is something the scenario *makes*
/// rather than something the fixture contains.
pub(crate) fn two_tenant_state() -> DesiredState {
    let mut state = fixtures::two_tenant_directory_state();
    let credential = fixtures::credential(&other(), 13, "secondary");
    state
        .insert(credential.clone())
        .and_then(|state| {
            state.insert(fixtures::alias(
                &other(),
                14,
                "steady",
                &[credential.reference],
            ))
        })
        .and_then(|state| state.insert(fixtures::tenant_policy(1, 1)))
        .and_then(|state| state.insert(fixtures::tenant_policy(11, 1)))
        .expect("two tenants that reference nothing of each other's are valid");
    state
}

/// [`two_tenant_state`] with the neighbour left out, and otherwise identical:
/// the caller's tenant, project, directory and policy, and nothing else.
///
/// The comparison half of a "a neighbour changes nothing" scenario. It has to be
/// this state minus the neighbour rather than some other single-tenant fixture,
/// or the two projections differ by more than the neighbour and the assertion
/// stops being about isolation the moment either fixture grows a field the
/// projection reads.
pub(crate) fn one_tenant_state() -> DesiredState {
    let mut state = fixtures::state_with_directory();
    state
        .insert(fixtures::tenant_policy(1, 1))
        .expect("one tenant's own policy over its own tenant is valid");
    state
}

/// [`two_tenant_state`] plus each tenant's own catalogue: one tenant-wide
/// enablement and one typed project alias resolving to it, per tenant.
///
/// Both tenants enable *the same offering* from the same deployment-wide
/// catalogue snapshot, which is the case worth asserting: a lookup keyed on the
/// offering alone, or on the alias slug alone, would answer one tenant's question
/// with the other's row. Distinct slugs, so a slug collision is not what the
/// scenarios are measuring.
pub(crate) fn two_tenant_catalogue_state() -> DesiredState {
    let mut state = two_tenant_state();
    let mine = fixtures::tenant_enablement(&caller(), 50, MODEL);
    let theirs = fixtures::tenant_enablement(&other(), 60, MODEL);
    state
        .insert(mine.clone())
        .and_then(|state| {
            state.insert(fixtures::typed_alias(
                &caller(),
                &fixtures::project_id(2),
                51,
                "quick",
                &[mine.reference],
            ))
        })
        .and_then(|state| state.insert(theirs.clone()))
        .and_then(|state| {
            state.insert(fixtures::typed_alias(
                &other(),
                &fixtures::project_id(12),
                61,
                "swift",
                &[theirs.reference],
            ))
        })
        .expect("each tenant enabling the same offering for itself is valid");
    state
}

/// The offering both tenants enable in [`two_tenant_catalogue_state`].
pub(crate) const MODEL: &str = "gpt-4o";

/// A real control-plane journal on a schema of its own.
pub(crate) struct Journal {
    pub(crate) store: Arc<PostgresControlPlane>,
    dsn: String,
    schema: String,
    /// The login roles this fixture created, so [`Drop`] can remove them.
    roles: Mutex<Vec<String>>,
    /// The password this fixture's roles were created with, generated per run.
    password: String,
    /// The claim on [`Journal::schema`]. Declared last, so it is dropped after
    /// this fixture's own [`Drop`] has removed the roles that were granted on
    /// the schema it takes.
    _schema: SchemaClaim,
}

impl Journal {
    /// A migrated journal, or `None` when no Postgres is configured and the
    /// suite is not running in required mode.
    ///
    /// `AXOND_TEST_REQUIRE_SERVICES=1` turns the `None` into a panic
    /// ([`crate::test_services`]), so the stateful lane cannot report green by
    /// skipping every scenario in this module family.
    pub(crate) async fn open() -> Option<Self> {
        let dsn = crate::test_services::postgres_dsn()?;
        let schema = format!(
            "ti_{}",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("a monotonic wall clock")
                .as_nanos()
        );
        // Claimed before it exists, because the migration below is exactly the
        // step that can fail: a scenario whose journal refuses to migrate would
        // otherwise leave its schema on a shared database with nothing owning
        // the `DROP`.
        let claim = SchemaClaim::create(&dsn, &schema).await;
        let store = PostgresControlPlane::connect(
            &dsn,
            ControlPlaneSettings {
                schema: Some(schema.clone()),
                operation_timeout: Duration::from_secs(10),
                connect_timeout: Duration::from_secs(5),
                ..ControlPlaneSettings::default()
            },
        )
        .await
        .expect("a migrated journal");
        Some(Self {
            store: Arc::new(store),
            dsn,
            schema,
            roles: Mutex::new(Vec::new()),
            password: role_password(),
            _schema: claim,
        })
    }

    /// The store as the administrative service holds it.
    pub(crate) fn store(&self) -> Arc<dyn ControlPlaneStore> {
        self.store.clone()
    }

    pub(crate) fn schema(&self) -> &str {
        &self.schema
    }

    /// Publish `state` as the next revision, attributed by `key`.
    pub(crate) async fn publish(
        &self,
        key: &str,
        expected: ExpectedRevision,
        state: DesiredState,
    ) -> Result<RevisionId, ControlPlaneError> {
        self.store
            .publish_revision(fixtures::candidate(expected, key, state))
            .await
            .map(|manifest| manifest.id)
    }

    /// Publish [`two_tenant_state`] as the first revision.
    pub(crate) async fn publish_two_tenants(&self) -> RevisionId {
        self.publish("two-tenants", ExpectedRevision::Empty, two_tenant_state())
            .await
            .expect("two tenants that reference nothing of each other's publish")
    }

    pub(crate) async fn head(&self) -> Option<RevisionId> {
        self.store
            .desired_revision()
            .await
            .expect("the head is readable")
    }

    /// The head revision, hydrated as a replica hydrates it.
    pub(crate) async fn hydrated(&self) -> LoadedRevision {
        self.store
            .load_desired_revision()
            .await
            .expect("the head hydrates")
            .expect("a published head")
    }

    /// A session as an ordinary application role: `privileges` on every table in
    /// the schema, pinned to `tenant` when one is given.
    ///
    /// An unpinned session (`None`) is the publisher: `axond.tenant_id` is
    /// unset, every policy admits everything, and it is what the scenarios read
    /// through to prove a row a pinned session could not see is nevertheless
    /// still there.
    pub(crate) async fn session(
        &self,
        label: &str,
        privileges: &str,
        tenant: Option<TenantId>,
    ) -> Client {
        let role = format!("{}_{label}", self.schema);
        let schema = &self.schema;
        let password = &self.password;
        connect(&self.dsn)
            .await
            .batch_execute(&format!(
                "CREATE ROLE {role} LOGIN PASSWORD '{password}'; \
                 GRANT USAGE ON SCHEMA {schema} TO {role}; \
                 GRANT {privileges} ON ALL TABLES IN SCHEMA {schema} TO {role}"
            ))
            .await
            .expect("a scenario role");
        self.roles.lock().expect("the role list").push(role.clone());

        let client = connect_as(&self.dsn, &role, password).await;
        let pin = match tenant {
            Some(tenant) => format!("SET axond.tenant_id = '{tenant}'"),
            None => String::from("RESET axond.tenant_id"),
        };
        client
            .batch_execute(&format!("SET search_path TO {schema}; {pin}"))
            .await
            .expect("a pinned session");
        client
    }

    /// One text column of a query, run as the migrating role: what is *actually*
    /// stored, whatever a pinned session can see of it.
    pub(crate) async fn stored(&self, sql: &str) -> Vec<String> {
        let client = connect(&self.dsn).await;
        client
            .batch_execute(&format!("SET search_path TO {}", self.schema))
            .await
            .expect("the journal's schema");
        column(&client, sql).await
    }
}

impl Drop for Journal {
    /// Drop the schema and every role this scenario created, even when the
    /// scenario panicked.
    ///
    /// On its own thread with its own runtime, because [`Drop`] is synchronous
    /// and the surrounding runtime may already be unwinding. A cleanup that
    /// itself fails is a panic *unless* the test is already panicking, where a
    /// second panic would replace the assertion failure the operator needs to
    /// read with a teardown error.
    ///
    /// The roles go first and every statement is attempted whatever the ones
    /// before it did: a role is a credential on a shared database, and a schema
    /// drop that fails transiently must not be what decides whether the
    /// credential outlives the test. Failures are collected and reported
    /// together, so one leak does not hide another. The schema itself is taken
    /// after this body by [`SchemaClaim`], which has owned it since before it
    /// existed.
    fn drop(&mut self) {
        let dsn = self.dsn.clone();
        let roles = std::mem::take(&mut *self.roles.lock().expect("the role list"));
        let cleanup = std::thread::spawn(move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("a cleanup runtime");
            runtime.block_on(async {
                let client = connect(&dsn).await;
                let mut left_behind = Vec::new();
                for role in roles {
                    if let Err(error) = client
                        .batch_execute(&format!(
                            "DROP OWNED BY {role} CASCADE; DROP ROLE IF EXISTS {role}"
                        ))
                        .await
                    {
                        left_behind.push(format!("the login role {role}: {}", detail(&error)));
                    }
                }
                assert!(
                    left_behind.is_empty(),
                    "a scenario could not clean up after itself: {left_behind:?}"
                );
            });
        });
        if cleanup.join().is_err() && !std::thread::panicking() {
            panic!("a scenario left its login roles behind");
        }
    }
}

/// The claim on a scenario's schema, held from before the `CREATE` that makes
/// it: whatever fails afterwards — the migration, an assertion, the fixture's
/// own construction — the `DROP` has an owner that outlives the failure.
struct SchemaClaim {
    dsn: String,
    schema: String,
}

impl SchemaClaim {
    /// Claim `schema` and create it, in that order: a `CREATE` that half
    /// succeeded is a schema too, and it is claimed already.
    async fn create(dsn: &str, schema: &str) -> Self {
        let claimed = Self {
            dsn: dsn.to_owned(),
            schema: schema.to_owned(),
        };
        connect(dsn)
            .await
            .batch_execute(&format!("CREATE SCHEMA {schema}"))
            .await
            .expect("a fresh scenario schema");
        claimed
    }
}

impl Drop for SchemaClaim {
    /// On a thread of its own with its own runtime, for the same reason
    /// [`Journal`]'s is: [`Drop`] cannot await, and the surrounding runtime may
    /// already be unwinding.
    fn drop(&mut self) {
        let dsn = self.dsn.clone();
        let schema = self.schema.clone();
        let cleanup = std::thread::spawn(move || {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("a cleanup runtime")
                .block_on(async move {
                    connect(&dsn)
                        .await
                        .batch_execute(&format!("DROP SCHEMA IF EXISTS {schema} CASCADE"))
                        .await
                        .expect("the scenario's schema is dropped");
                });
        });
        if cleanup.join().is_err() && !std::thread::panicking() {
            panic!("a scenario left its schema {} behind", self.schema);
        }
    }
}

/// A connection to the test database as the DSN's own user.
async fn connect(dsn: &str) -> Client {
    open(dsn.parse().expect("a parseable test DSN")).await
}

/// A connection to the test database as one of a fixture's scenario roles.
async fn connect_as(dsn: &str, role: &str, password: &str) -> Client {
    let mut config: Config = dsn.parse().expect("a parseable test DSN");
    config.user(role).password(password);
    open(config).await
}

async fn open(mut config: Config) -> Client {
    config.connect_timeout(Duration::from_secs(5));
    let (client, connection) = config
        .connect(crate::usage::tls_connector())
        .await
        .expect("a connection to the test database");
    tokio::spawn(async move {
        let _ = connection.await;
    });
    client
}

/// A database error with the server's own message, which
/// [`tokio_postgres::Error`]'s own rendering keeps in its source rather than its
/// `Display`. An assertion about *why* a write was refused is worthless if the
/// text it matches on is the constant `db error`.
pub(crate) fn detail(error: &tokio_postgres::Error) -> String {
    let mut rendered = error.to_string();
    let mut source = std::error::Error::source(error);
    while let Some(cause) = source {
        rendered.push_str(": ");
        rendered.push_str(&cause.to_string());
        source = cause.source();
    }
    rendered
}

/// The first column of `sql`, as text, with `NULL` rendered rather than dropped.
pub(crate) async fn column(client: &Client, sql: &str) -> Vec<String> {
    client
        .query(sql, &[])
        .await
        .unwrap_or_else(|error| panic!("the read itself must succeed: {}", detail(&error)))
        .iter()
        .map(|row| {
            row.try_get::<_, Option<String>>(0)
                .expect("a text column")
                .unwrap_or_else(|| "<null>".to_owned())
        })
        .collect()
}

/// Whether `sql` is refused by the database, and how.
pub(crate) async fn refused(client: &Client, sql: &str) -> String {
    let refusal = client
        .batch_execute(sql)
        .await
        .expect_err("the database must refuse the write");
    detail(&refusal)
}

/// How many rows `sql` affected. Zero is the answer row-level security gives a
/// pinned `UPDATE` or `DELETE` that names another tenant's rows: they are not
/// there to be matched.
pub(crate) async fn affected(client: &Client, sql: &str) -> u64 {
    client
        .execute(sql, &[])
        .await
        .unwrap_or_else(|error| panic!("the statement itself must run: {}", detail(&error)))
}

/// Identifiers that must not appear in a surface a caller can see.
///
/// Exact matches, per this module's header: the fixtures' ids are seeded, so a
/// fragment search would flag a tenant's own id as its neighbour's.
pub(crate) struct Absent {
    names: Vec<(&'static str, String)>,
}

impl Absent {
    pub(crate) fn of(names: impl IntoIterator<Item = (&'static str, String)>) -> Self {
        let names: Vec<_> = names.into_iter().collect();
        assert!(
            names.iter().all(|(_, value)| !value.is_empty()),
            "an empty identifier would make every absence assertion vacuous"
        );
        Self { names }
    }

    /// Every identifier of the tenant a scenario must not reach: its id, its
    /// slug, its project, its administrator, its workload, its credential and
    /// the secret that credential points at.
    pub(crate) fn of_the_other_tenant() -> Self {
        let credential = fixtures::credential(&other(), 13, "secondary");
        Self::of([
            ("tenant id", other().to_string()),
            ("tenant slug", "globex".to_owned()),
            ("project id", fixtures::project_id(12).to_string()),
            ("principal id", fixtures::principal_id(40).to_string()),
            ("workload id", fixtures::principal_id(41).to_string()),
            ("credential id", credential.reference.id.to_string()),
            ("secret id", fixtures::secret_id(13).to_string()),
        ])
    }

    /// The same identifiers minus the other tenant's registration — its tenant id
    /// and its slug — which the second wall does not claim to hide.
    ///
    /// A tenant *resource* is deployment-scoped: it is the row that declares a
    /// tenant exists, `tenant_id` is `NULL` on it, and every policy in
    /// `control_plane_0002_tenancy_access.sql` admits a `NULL` owner because the
    /// journal is deployment-wide history. So a pinned session reading
    /// `axond_cp_resource_version` can enumerate which tenants exist and what they
    /// are called, and only the service layer refuses that read. Named here rather
    /// than swept for, because an assertion that quietly excluded it would hide a
    /// real surface: [`super::database`] asserts the visibility instead of the
    /// absence, so the day the policy changes, the test that changes is this one.
    pub(crate) fn of_the_other_tenants_own_rows() -> Self {
        let all = Self::of_the_other_tenant();
        Self {
            names: all
                .names
                .into_iter()
                .filter(|(label, _)| !matches!(*label, "tenant id" | "tenant slug"))
                .collect(),
        }
    }

    /// The identifiers, so a scenario can assert the other direction: that what it
    /// is looking for is stored at all, and absence is therefore the wall's doing.
    pub(crate) fn names(&self) -> &[(&'static str, String)] {
        &self.names
    }

    /// Assert that no identifier appears in `rendered`.
    ///
    /// `surface` names what was read — "the refusal the caller was given", "the
    /// pinned session's projected rows" — because that is what makes a failure
    /// actionable.
    pub(crate) fn assert_absent(&self, surface: &str, rendered: &str) {
        for (label, value) in &self.names {
            assert!(
                !rendered.contains(value.as_str()),
                "{surface} discloses the other tenant's {label}: {rendered}"
            );
        }
    }
}

#[cfg(test)]
mod claim {
    use super::{SchemaClaim, connect};

    /// Whether `schema` exists on the test database.
    async fn exists(dsn: &str, schema: &str) -> bool {
        connect(dsn)
            .await
            .query_one(
                "SELECT count(*) FROM pg_namespace WHERE nspname = $1",
                &[&schema],
            )
            .await
            .expect("a schema lookup")
            .get::<_, i64>(0)
            > 0
    }

    /// The window the claim exists for: a setup step that fails after the
    /// `CREATE SCHEMA` and before the fixture is built still takes the schema
    /// with it, so a long-lived CI database does not accumulate one abandoned
    /// schema per failed scenario.
    #[tokio::test]
    async fn a_setup_that_fails_after_the_create_still_drops_the_schema() {
        let Some(dsn) = crate::test_services::postgres_dsn() else {
            return;
        };
        let schema = format!("ti_claim_{}", std::process::id());
        /// The arranged failure, so the case cannot pass on some other panic.
        const ARRANGED: &str = "a scenario's setup failed after its schema existed";

        // A thread with a runtime of its own: the failure has to unwind without
        // taking this test with it, and the destructor's cleanup cannot re-enter
        // the runtime it is unwinding.
        let outcome = std::thread::spawn({
            let (dsn, schema) = (dsn.clone(), schema.clone());
            move || {
                tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("a runtime for the failing setup")
                    .block_on(async move {
                        let claimed = SchemaClaim::create(&dsn, &schema).await;
                        assert!(
                            exists(&dsn, &claimed.schema).await,
                            "the arranged setup created its schema, or the case proves nothing"
                        );
                        // Still a live local, so it is the unwind that has to
                        // clean up — which is the property under test.
                        panic!("{ARRANGED}");
                    });
            }
        })
        .join();

        let Err(panic) = outcome else {
            panic!("the arranged setup returned instead of failing");
        };
        let message = panic
            .downcast_ref::<String>()
            .map(String::as_str)
            .or_else(|| panic.downcast_ref::<&str>().copied())
            .unwrap_or_default();
        assert!(
            message.contains(ARRANGED),
            "the setup failed for the arranged reason, not another: {message}"
        );
        assert!(
            !exists(&dsn, &schema).await,
            "the failed setup left the schema {schema} behind"
        );
    }
}