pg_tviews 0.1.0-beta.32

pub mod collections;
pub mod graph;
pub(crate) mod indexes;
pub mod plan;
pub mod queries;
pub mod reads;
pub mod registered;
pub mod resolve;
pub mod row;

pub use collections::Collection;
pub use graph::EntityDepGraph;
use pgrx::pg_sys::Oid;
use pgrx::prelude::*;
use std::rc::Rc;

/// The catalog table, schema-qualified: every read or write of it names it so.
#[must_use]
pub fn meta_table() -> String {
    format!("{}.pg_tview_meta", crate::utils::ext_schema())
}

/// Represents a row in `pg_tview_meta` (your own catalog table).
#[derive(Debug, Clone)]
pub struct TviewMeta {
    pub tview_oid: Oid,
    pub view_oid: Oid,
    pub entity_name: String,

    /// What registration derived from the backing view's query tree (ADR 0203).
    pub plan: plan::TviewPlan,

    /// What a write to a base table no cascade maps to its keys does: the policy
    /// stored when the TVIEW was created.
    pub uncascaded_policy: crate::config::UncascadedPolicy,

    /// The tables declared with a policy of their own, each overriding
    /// `uncascaded_policy` for writes to it.
    pub table_policies: Vec<(Oid, crate::config::UncascadedPolicy)>,

    /// The functions the definition calls, declared with the tables each reads:
    /// `schema.name(argument types)`.
    pub function_reads: Vec<(String, Vec<Oid>)>,

    /// A TVIEW that reads the current time declared `time_refresh: external`.
    pub time_refresh_external: bool,
    /// The definition reads the current time: two computations of a row may
    /// differ with no write between them.
    pub time_dependent: bool,

    /// The column that names the TVIEW's rows (ADR 0169).
    pub identity: RowIdentity,

    /// The GraphQL type name of its rows ([`typename`]), shared by every change
    /// the journal records.
    pub typename: std::rc::Rc<str>,
    /// How its table's `id` column, the rows' public id that the journal carries,
    /// is written in JSON; `None` without such a column.
    pub public_id: Option<PublicId>,
    /// Its references to other TVIEWs, as the change feed names the collections
    /// its rows belong to.
    pub collections: Vec<Rc<Collection>>,
}

/// A TVIEW's entity and plan: all a write reads of the TVIEWs it does not reach
/// (the cascade paths starting at the written table, the dependency graph).
#[derive(Debug, Clone, Default)]
pub struct PlanEntry {
    /// The TVIEW's entity.
    pub entity: String,
    /// Its propagation plan.
    pub plan: plan::TviewPlan,
}

/// How a TVIEW's public `id` is written in JSON (as `to_jsonb` writes it).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PublicId {
    /// An integer or numeric: its text is the JSON number.
    Number,
    /// Anything else (a uuid): a JSON string.
    String,
}

/// A row's public id as the write that changed it returned it
/// ([`TviewMeta::id_text`]): kept as text, written in JSON only where it is
/// reported or published.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RowId {
    pub text: String,
    pub kind: PublicId,
}

impl RowId {
    /// The id as JSON text, as `to_jsonb` writes it.
    #[must_use]
    pub fn json(&self) -> String {
        match self.kind {
            PublicId::Number => self.text.clone(),
            PublicId::String => serde_json::Value::String(self.text.clone()).to_string(),
        }
    }

    /// The id as a JSON value.
    #[must_use]
    pub fn value(&self) -> serde_json::Value {
        match self.kind {
            PublicId::Number => serde_json::from_str(&self.text)
                .unwrap_or_else(|_| serde_json::Value::String(self.text.clone())),
            PublicId::String => serde_json::Value::String(self.text.clone()),
        }
    }
}

/// The column that names a TVIEW's rows (ADR 0169), as the catalog records it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RowIdentity {
    pub column: String,
    pub kind: crate::lineage::IdentityKind,
}

/// How the values of an identity column are carried and bound.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KeyType {
    /// `int2`, `int4` or `int8`: bound as `int8`.
    Int,
    /// Any other type, schema-qualified and quoted: bound as text, cast to it.
    Text(String),
}

impl RowIdentity {
    /// Read `pg_tview_meta.identity`, which every registration writes.
    ///
    /// # Errors
    /// It is NULL, names no column, or has an unknown kind.
    pub fn from_catalog(
        entity: &str,
        json: Option<&serde_json::Value>,
    ) -> crate::TViewResult<Self> {
        let unreadable = |why: &str| crate::TViewError::CatalogError {
            operation: format!("Read the row identity of tv_{entity}"),
            pg_error: why.to_string(),
        };
        let json = json.ok_or_else(|| unreadable("it is NULL"))?;
        let column = json["columns"][0]["name"]
            .as_str()
            .ok_or_else(|| unreadable("it names no column"))?
            .to_string();
        let kind = match json["kind"].as_str() {
            Some("pk") => crate::lineage::IdentityKind::Pk,
            Some("distinct_on") => crate::lineage::IdentityKind::DistinctOn,
            other => return Err(unreadable(&format!("unknown kind {other:?}"))),
        };
        Ok(Self { column, kind })
    }

    /// No identity: what a catalog row awaiting re-registration has.
    const fn unknown() -> Self {
        Self {
            column: String::new(),
            kind: crate::lineage::IdentityKind::Pk,
        }
    }

    /// Whether the identity is `pk_<entity>`, the column parents join on.
    #[must_use]
    pub fn is_pk(&self, entity: &str) -> bool {
        self.column == format!("pk_{entity}")
    }
}

/// Shared SELECT column list + FROM used by every `TviewMeta` loader. Callers
/// append their own `WHERE` / `ORDER BY`. One copy keeps the loaders from drifting
/// out of sync as catalog columns are added.
///
/// Inside the extension's own script (an upgrade re-registering TVIEWs between two
/// scripts) the catalog can be an earlier release's, without the view the
/// collections are read from: they are left out, and nothing loaded then is
/// cached ([`remember`]).
pub(crate) fn meta_select() -> String {
    let collections = if crate::revision::in_own_script() {
        "NULL::pg_catalog.jsonb AS collections".to_string()
    } else {
        collections::collections_sql()
    };
    format!(
        "SELECT table_oid::oid AS tview_oid, view_oid::oid AS view_oid, entity, plan, \
         uncascaded_policy, identity, \
         uncascaded_table_oids::oid[] AS uncascaded_table_oids, uncascaded_table_policies, \
         function_read_functions, function_read_tables::oid[] AS function_read_tables, \
         time_refresh IS NOT DISTINCT FROM 'external' AS time_refresh_external, \
         time_dependent, graphql_typename, \
         (SELECT a.atttypid IN ('pg_catalog.int2'::pg_catalog.regtype, \
                                'pg_catalog.int4'::pg_catalog.regtype, \
                                'pg_catalog.int8'::pg_catalog.regtype, \
                                'pg_catalog.numeric'::pg_catalog.regtype) \
          FROM pg_catalog.pg_attribute a WHERE a.attrelid = table_oid \
          AND a.attname = 'id' AND a.attnum > 0 AND NOT a.attisdropped) AS public_id_number, \
         {collections} \
         FROM {}",
        crate::catalog::meta_table()
    )
}

/// The cached catalog row of `entity`, unless `pg_tviews.table_cache_enabled` is
/// off.
fn cached(entity: &str) -> Option<Rc<TviewMeta>> {
    if !crate::config::table_cache_enabled() {
        return None;
    }
    let meta = crate::cache::METAS.with(|m| m.get(&entity.to_string()));
    if meta.is_some() {
        crate::metrics::metrics_api::record_table_cache_hit();
    } else {
        crate::metrics::metrics_api::record_table_cache_miss();
    }
    meta
}

/// Count the catalog query just made and cache its result, unless the extension's
/// own script is running ([`meta_select`]).
fn remember(loaded: Option<TviewMeta>) -> Option<Rc<TviewMeta>> {
    crate::metrics::metrics_api::record_catalog_lookup();
    let loaded = loaded.map(Rc::new);
    if crate::revision::in_own_script() {
        return loaded;
    }
    if let Some(meta) = &loaded {
        crate::cache::watch(&[meta.tview_oid, meta.view_oid]);
        // A parent's `id` column is part of how the collections are named.
        let parents: Vec<Oid> = meta
            .collections
            .iter()
            .map(|c| c.parent_table_oid)
            .collect();
        crate::cache::watch(&parents);
        crate::cache::METAS.with(|m| m.insert(meta.entity_name.clone(), meta.clone()));
    }
    loaded
}

impl TviewMeta {
    /// The rows' public id as text, read from `table`, its table's name or alias
    /// in a statement; `NULL` for a TVIEW without an `id` column.
    #[must_use]
    pub fn id_text(&self, table: &str) -> String {
        if self.public_id.is_some() {
            format!("{table}.\"id\"::pg_catalog.text")
        } else {
            "NULL::pg_catalog.text".to_string()
        }
    }

    /// The public id `text` a write returned ([`Self::id_text`]), with its kind.
    #[must_use]
    pub fn row_id(&self, text: String) -> RowId {
        RowId {
            text,
            kind: self.public_id.unwrap_or(PublicId::String),
        }
    }

    /// How the values of the identity column are bound: the type of the column in
    /// the TVIEW's table (cached per backend).
    ///
    /// # Errors
    /// Returns an error if the catalog cannot be read.
    pub fn key_type(&self) -> crate::TViewResult<KeyType> {
        if let Some(key_type) = crate::cache::KEY_TYPES.with(|m| m.get(&self.tview_oid)) {
            return Ok(key_type);
        }
        let name = Spi::get_one_with_args::<String>(
            "SELECT CASE WHEN a.atttypid IN ('pg_catalog.int2'::pg_catalog.regtype, \
                                             'pg_catalog.int4'::pg_catalog.regtype, \
                                             'pg_catalog.int8'::pg_catalog.regtype) THEN NULL \
                    ELSE pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(t.typname) END \
             FROM pg_catalog.pg_attribute a \
             JOIN pg_catalog.pg_type t ON t.oid = a.atttypid \
             JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace \
             WHERE a.attrelid = $1 AND a.attname = $2 AND NOT a.attisdropped",
            &[
                crate::utils::spi::oid(self.tview_oid),
                crate::utils::spi::text(self.identity.column.as_str()),
            ],
        )
        .or_else(|e| match e {
            spi::Error::InvalidPosition => Ok(None),
            e => Err(e),
        })?;
        let key_type = name.map_or(KeyType::Int, KeyType::Text);
        crate::cache::KEY_TYPES.with(|m| m.insert(self.tview_oid, key_type.clone()));
        Ok(key_type)
    }

    /// What a write to `table_oid`, a table no cascade reaches, does: its own
    /// declared policy, else the TVIEW's.
    #[must_use]
    pub fn policy_for(&self, table_oid: Oid) -> crate::config::UncascadedPolicy {
        self.table_policies
            .iter()
            .find(|(oid, _)| *oid == table_oid)
            .map_or(self.uncascaded_policy, |(_, policy)| *policy)
    }

    /// The mapping of base table `table_oid`, also found by name (a row restored
    /// before its relids were rebound) or through the partitioned table `root`.
    #[must_use]
    pub fn key_mapping(
        &self,
        table_oid: Oid,
        root: Option<Oid>,
    ) -> Option<&crate::lineage::KeyMapping> {
        self.plan
            .tables
            .iter()
            .find(|m| m.relid == table_oid.to_u32() || root.is_some_and(|r| m.relid == r.to_u32()))
    }

    /// Look up metadata by entity name (cached per backend, shared: a lookup
    /// copies nothing).
    pub fn load_by_entity(entity_name: &str) -> crate::TViewResult<Option<Rc<Self>>> {
        if let Some(meta) = cached(entity_name) {
            return Ok(Some(meta));
        }
        let loaded = Spi::connect(|client| -> crate::TViewResult<Option<Self>> {
            let args = vec![crate::utils::spi::text(entity_name)];
            let mut rows =
                client.select(&format!("{} WHERE entity = $1", meta_select()), None, &args)?;

            match rows.next() {
                Some(row) => Ok(Some(Self::from_spi_row(&row)?)),
                None => Ok(None),
            }
        })?;
        Ok(remember(loaded))
    }

    /// Load all TVIEW metadata
    pub fn load_all() -> crate::TViewResult<Vec<Self>> {
        Spi::connect(|client| -> crate::TViewResult<Vec<Self>> {
            let rows = client.select(&format!("{} ORDER BY entity", meta_select()), None, &[])?;

            let mut result = Vec::new();
            for row in rows {
                result.push(Self::from_spi_row(&row)?);
            }
            Ok(result)
        })
    }

    /// Every TVIEW's entity and plan, by entity: the write path's read of the
    /// whole catalog. It carries no per-TVIEW subquery and resolves no name, so a
    /// write does not depend on what the writer may see of the TVIEWs it does not
    /// reach (#251); their full rows are loaded by entity when a write reaches them.
    ///
    /// # Errors
    /// The catalog cannot be read, or a plan does not decode.
    pub fn load_plans() -> crate::TViewResult<Vec<PlanEntry>> {
        Spi::connect(|client| -> crate::TViewResult<Vec<PlanEntry>> {
            let rows = client.select(
                &format!(
                    "SELECT entity, plan FROM {} ORDER BY entity",
                    crate::catalog::meta_table()
                ),
                None,
                &[],
            )?;
            let mut entries = Vec::new();
            for row in rows {
                let entity = row["entity"].value::<String>()?.unwrap_or_default();
                let plan = plan::TviewPlan::decode(
                    &entity,
                    row["plan"]
                        .value::<pgrx::JsonB>()?
                        .map_or(serde_json::Value::Null, |j| j.0),
                )?;
                entries.push(PlanEntry { entity, plan });
            }
            Ok(entries)
        })
    }

    /// The TVIEWs whose catalog row does not decode, each with why.
    ///
    /// # Errors
    /// The catalog cannot be read.
    pub fn unreadable() -> crate::TViewResult<Vec<(String, String)>> {
        Spi::connect(|client| -> crate::TViewResult<Vec<(String, String)>> {
            let rows = client.select(&format!("{} ORDER BY entity", meta_select()), None, &[])?;
            let mut unreadable = Vec::new();
            for row in rows {
                if let Err(e) = Self::from_spi_row(&row) {
                    let entity = row["entity"].value::<String>()?.unwrap_or_default();
                    unreadable.push((entity, e.to_string()));
                }
            }
            Ok(unreadable)
        })
    }

    /// The entity of the TVIEW whose table is `tview_oid`, read without decoding
    /// its plan: what dropping a TVIEW needs.
    ///
    /// # Errors
    /// The catalog cannot be read. Raises the revision error for a catalog of
    /// another revision.
    pub fn entity_of_table(tview_oid: Oid) -> crate::TViewResult<Option<String>> {
        // The hooks' first catalog read in a statement: a catalog of another
        // revision fails here with its remedy, not on a table or column it lacks.
        crate::revision::check();
        crate::utils::spi::one::<String>(
            &format!(
                "SELECT entity FROM {} WHERE table_oid::pg_catalog.oid = $1",
                crate::catalog::meta_table()
            ),
            &[crate::utils::spi::oid(tview_oid)],
        )
    }

    /// The catalog row of `entity_name` for re-registration, which derives its
    /// plan and identity again: either one that does not decode reads as empty, so
    /// re-registering is the remedy for it. Never cached.
    ///
    /// # Errors
    /// The catalog cannot be read.
    pub fn load_to_rederive(entity_name: &str) -> crate::TViewResult<Option<Self>> {
        Spi::connect(|client| -> crate::TViewResult<Option<Self>> {
            let args = vec![crate::utils::spi::text(entity_name)];
            let mut rows =
                client.select(&format!("{} WHERE entity = $1", meta_select()), None, &args)?;
            rows.next()
                .map(|row| Self::decode(&row, Derived::Rederive))
                .transpose()
        })
    }

    /// Parse a row of [`meta_select`] into a `TviewMeta`.
    pub fn from_spi_row(row: &spi::SpiHeapTupleData) -> crate::TViewResult<Self> {
        Self::decode(row, Derived::Strict)
    }

    fn decode(row: &spi::SpiHeapTupleData, derived: Derived) -> crate::TViewResult<Self> {
        let entity_name: String =
            row["entity"]
                .value()?
                .ok_or_else(|| crate::TViewError::SpiError {
                    query: String::new(),
                    error: "entity column is NULL".to_string(),
                })?;
        let policy = |name: &str| {
            crate::config::UncascadedPolicy::parse(name).ok_or_else(|| {
                crate::TViewError::CatalogError {
                    operation: format!("Read the uncascaded policies of tv_{entity_name}"),
                    pg_error: format!("unknown policy {name:?}"),
                }
            })
        };
        let uncascaded_policy = policy(
            &row["uncascaded_policy"]
                .value::<String>()?
                .unwrap_or_default(),
        )?;

        let table_policies = row["uncascaded_table_oids"]
            .value::<Vec<Oid>>()?
            .unwrap_or_default()
            .into_iter()
            .zip(
                row["uncascaded_table_policies"]
                    .value::<Vec<String>>()?
                    .unwrap_or_default(),
            )
            .map(|(oid, name)| Ok((oid, policy(&name)?)))
            .collect::<crate::TViewResult<_>>()?;

        let mut function_reads: Vec<(String, Vec<Oid>)> = Vec::new();
        for (function, table) in row["function_read_functions"]
            .value::<Vec<String>>()?
            .unwrap_or_default()
            .into_iter()
            .zip(
                row["function_read_tables"]
                    .value::<Vec<Option<Oid>>>()?
                    .unwrap_or_default(),
            )
        {
            if function_reads.last().is_none_or(|(f, _)| *f != function) {
                function_reads.push((function, Vec::new()));
            }
            if let (Some(table), Some((_, tables))) = (table, function_reads.last_mut()) {
                tables.push(table);
            }
        }

        let identity = RowIdentity::from_catalog(
            &entity_name,
            row["identity"]
                .value::<pgrx::JsonB>()?
                .map(|j| j.0)
                .as_ref(),
        );
        let identity = match (identity, derived) {
            (Ok(identity), _) => identity,
            (Err(_), Derived::Rederive) => RowIdentity::unknown(),
            (Err(e), Derived::Strict) => return Err(e),
        };
        let plan = plan::TviewPlan::decode(
            &entity_name,
            row["plan"]
                .value::<pgrx::JsonB>()?
                .map_or(serde_json::Value::Null, |j| j.0),
        );
        let plan = match (plan, derived) {
            (Ok(plan), _) => plan,
            (Err(_), Derived::Rederive) => plan::TviewPlan::default(),
            (Err(e), Derived::Strict) => return Err(e),
        };

        let typename: Rc<str> =
            typename(&entity_name, row["graphql_typename"].value::<String>()?).into();
        let collections = collections::decode(
            &typename,
            row["collections"]
                .value::<pgrx::JsonB>()?
                .map(|j| j.0)
                .as_ref(),
        );
        Ok(Self {
            tview_oid: row["tview_oid"]
                .value()?
                .ok_or_else(|| crate::TViewError::SpiError {
                    query: String::new(),
                    error: "tview_oid column is NULL".to_string(),
                })?,
            view_oid: row["view_oid"]
                .value()?
                .ok_or_else(|| crate::TViewError::SpiError {
                    query: String::new(),
                    error: "view_oid column is NULL".to_string(),
                })?,
            entity_name,

            plan,
            uncascaded_policy,
            table_policies,
            function_reads,
            time_refresh_external: row["time_refresh_external"]
                .value::<bool>()?
                .unwrap_or(false),
            time_dependent: row["time_dependent"].value::<bool>()?.unwrap_or(false),
            identity,
            typename,
            public_id: row["public_id_number"].value::<bool>()?.map(|number| {
                if number {
                    PublicId::Number
                } else {
                    PublicId::String
                }
            }),
            collections,
        })
    }
}

/// How [`TviewMeta::decode`] treats what registration derived (the plan, the
/// identity) when it does not decode.
#[derive(Clone, Copy)]
enum Derived {
    /// An error naming the TVIEW.
    Strict,
    /// Empty: the caller derives it again.
    Rederive,
}

impl Default for TviewMeta {
    fn default() -> Self {
        Self {
            tview_oid: pg_sys::Oid::INVALID,
            view_oid: pg_sys::Oid::INVALID,
            entity_name: String::new(),

            plan: plan::TviewPlan::default(),
            uncascaded_policy: crate::config::UncascadedPolicy::Warn,
            table_policies: Vec::new(),
            function_reads: Vec::new(),
            time_refresh_external: false,
            time_dependent: false,
            identity: RowIdentity::unknown(),
            typename: "".into(),
            public_id: None,
            collections: Vec::new(),
        }
    }
}

/// The GraphQL type name of `entity`'s rows: the `typename` option when declared,
/// else the entity in Pascal case (`blog_post` → `BlogPost`), capitalising ASCII letters
/// only (GraphQL names are ASCII), so `tviews.registry.typename` derives the same name in
/// SQL whatever the database's locale. The name `pg_tviews_flush_and_report()` emits and
/// its `types` argument takes.
#[must_use]
pub fn typename(entity: &str, declared: Option<String>) -> String {
    declared.unwrap_or_else(|| {
        entity
            .split('_')
            .filter(|part| !part.is_empty())
            .map(|part| {
                let mut chars = part.chars();
                chars.next().map_or_else(String::new, |first| {
                    first.to_ascii_uppercase().to_string() + chars.as_str()
                })
            })
            .collect()
    })
}

#[cfg(test)]
mod tests {
    #[test]
    fn a_public_id_is_written_in_json_as_to_jsonb_writes_it() {
        let meta = |public_id| super::TviewMeta {
            public_id,
            ..super::TviewMeta::default()
        };
        let number = meta(Some(super::PublicId::Number));
        let string = meta(Some(super::PublicId::String));
        assert_eq!(number.row_id("42".into()).json(), "42");
        assert_eq!(number.row_id("42".into()).value(), serde_json::json!(42));
        assert_eq!(string.row_id("8f0c".into()).json(), r#""8f0c""#);
        assert_eq!(string.row_id(r#"a"b\c"#.into()).json(), r#""a\"b\\c""#);
        assert_eq!(
            string.row_id("8f0c".into()).value(),
            serde_json::json!("8f0c")
        );
        assert_eq!(meta(None).id_text("t"), "NULL::pg_catalog.text");
        assert_eq!(string.id_text("t"), r#"t."id"::pg_catalog.text"#);
    }

    #[test]
    fn typename_is_the_option_or_the_pascal_case_of_the_entity() {
        assert_eq!(super::typename("post", None), "Post");
        assert_eq!(super::typename("blog_post", None), "BlogPost");
        assert_eq!(super::typename("order_line_item", None), "OrderLineItem");
        assert_eq!(super::typename("post", Some("Article".into())), "Article");
    }

    #[test]
    fn typename_capitalises_ascii_letters_only_whatever_the_locale() {
        // GraphQL names are ASCII; tviews.registry.typename derives the same name in SQL,
        // which must not depend on the database's ctype (Turkish dotted I, German sharp s).
        assert_eq!(super::typename("item", None), "Item");
        assert_eq!(super::typename("été", None), "été");
        assert_eq!(super::typename("ßeta_x", None), "ßetaX");
        assert_eq!(super::typename("_post__line_", None), "PostLine");
    }

    use super::*;

    #[test]
    fn row_identity_reads_the_catalog() {
        use crate::lineage::IdentityKind;
        let doc =
            serde_json::json!({"kind": "distinct_on", "columns": [{"name": "id", "type": "uuid"}]});
        let id = RowIdentity::from_catalog("doc", Some(&doc)).unwrap();
        assert_eq!(id.column, "id");
        assert_eq!(id.kind, IdentityKind::DistinctOn);
        assert!(!id.is_pk("doc"));

        let pk = serde_json::json!({"kind": "pk", "columns": [{"name": "pk_doc"}]});
        assert!(
            RowIdentity::from_catalog("doc", Some(&pk))
                .unwrap()
                .is_pk("doc")
        );
    }

    #[test]
    fn an_unreadable_row_identity_is_a_catalog_error() {
        let unknown = serde_json::json!({"kind": "natural", "columns": [{"name": "id"}]});
        for json in [
            None,
            Some(&serde_json::json!({"kind": "pk"})),
            Some(&unknown),
        ] {
            let err = RowIdentity::from_catalog("doc", json).unwrap_err();
            assert!(err.to_string().contains("tv_doc"), "{err}");
        }
    }
}