pg_tviews 0.1.0-beta.30

//! `pg_tviews_flush_and_report()`: the read-model rows a transaction changed, in
//! the GraphQL Cascade shape.
//!
//! A mutation function calls it last. It flushes whatever is still queued, then
//! reports every TVIEW row the transaction's refreshes inserted, updated or deleted
//! so far (the journal in [`crate::queue::affected`]), with each row's public `id`
//! and, optionally, its fresh `data`.

use crate::error::{TViewError, TViewResult};
use crate::queue::affected::{self, Change, NetChange};
use crate::utils::ident;
use crate::utils::spi::LookupKeys;
use serde_json::{Map, Value, json};
use std::collections::{BTreeSet, HashMap};

/// Flush pending refreshes and report the TVIEW rows this transaction changed.
///
/// Returns `{"updated": [...], "deleted": [...], "truncated": bool,
/// "invalidated_types": [...]}`. An updated entry is
/// `{"__typename", "id", "operation": "CREATED" | "UPDATED", "data"}` (`data` only
/// with `include_data`); a deleted entry is `{"__typename", "id"}`. Entries are
/// ordered by entity, then primary key. Past `max_entities` entries, or when the
/// journal overflowed `pg_tviews.report_max_tracked`, `truncated` is true and
/// `invalidated_types` lists the types of the rows left out. With `reset` (the
/// default) the next call reports only what changed after this one.
///
/// With `types` (GraphQL type names, [`crate::catalog::typename`]), only rows of
/// those types are reported and counted against `max_entities`; the others are
/// left out and their types listed in `omitted_types` (#230). `reset` clears them
/// too: they were seen, just not asked for.
///
/// # Errors
/// `InvalidInput` (22023) for a type name no TVIEW has, before anything is
/// flushed or reset; otherwise an error if the flush or a catalog/TVIEW read fails.
pub(crate) fn flush_and_report(
    max_entities: i32,
    include_data: bool,
    reset: bool,
    types: Option<Vec<String>>,
) -> TViewResult<Value> {
    let wanted = types.map(requested_types).transpose()?;
    crate::flush::flush_refresh_queue()?;
    let (changes, overflow) = affected::summarize(reset);
    let limit = usize::try_from(max_entities).unwrap_or(0);
    build_report(&changes, &overflow, limit, include_data, wanted.as_ref())
}

/// How many known type names an unknown-type error lists.
const KNOWN_TYPES_LISTED: usize = 20;

/// `types` as a set, once each is known to be a TVIEW's type name.
fn requested_types(types: Vec<String>) -> TViewResult<BTreeSet<String>> {
    let known: BTreeSet<String> = crate::catalog::row::typenames()?
        .into_iter()
        .map(|(_, typename)| typename)
        .collect();
    let unknown: Vec<&str> = types
        .iter()
        .filter(|t| !known.contains(*t))
        .map(String::as_str)
        .collect();
    if !unknown.is_empty() {
        let listed: Vec<&str> = known
            .iter()
            .take(KNOWN_TYPES_LISTED)
            .map(String::as_str)
            .collect();
        let more = if known.len() > KNOWN_TYPES_LISTED {
            ", …"
        } else {
            ""
        };
        return Err(TViewError::InvalidInput {
            parameter: "types".to_string(),
            reason: format!(
                "unknown type(s) {}; known: {}{more}",
                unknown.join(", "),
                listed.join(", ")
            ),
        });
    }
    Ok(types.into_iter().collect())
}

/// Per-entity type name and qualified TVIEW table.
struct EntityInfo {
    typename: String,
    table: String,
}

fn entity_info(entities: &BTreeSet<&str>) -> TViewResult<HashMap<String, EntityInfo>> {
    let names: Vec<String> = entities.iter().map(|e| (*e).to_string()).collect();
    Ok(crate::catalog::queries::typenames_and_tables(names)?
        .into_iter()
        .map(|(entity, typename, table)| (entity, EntityInfo { typename, table }))
        .collect())
}

/// The current `data` of `entity`'s rows `pks`, keyed by pk text.
///
/// `pk_<entity>` is BIGINT (#209) and indexed (its primary key, or its own index on a
/// TVIEW named by another key): the keys are bound as `int8` and each row is one index
/// probe, from a kept plan shaped by their count ([`LookupKeys`]), not a scan (#240).
///
/// # Errors
/// A journaled key that is not an integer, or a failed read.
fn current_data(entity: &str, table: &str, pks: &[String]) -> TViewResult<HashMap<String, Value>> {
    let keys = pks
        .iter()
        .map(|pk| {
            pk.parse::<i64>().map_err(|_| TViewError::CatalogError {
                operation: format!("Read the reported rows of tv_{entity}"),
                pg_error: format!("journaled key {pk:?} is not an integer"),
            })
        })
        .collect::<TViewResult<Vec<i64>>>()?;
    let Some(lookup) = LookupKeys::of(keys) else {
        return Ok(HashMap::new());
    };
    let qi_pk = ident::quoted(&format!("pk_{entity}"));
    let sql = format!(
        "SELECT t.{qi_pk}::pg_catalog.text, t.data::pg_catalog.text FROM {table} t WHERE {}",
        lookup.predicate(&format!("t.{qi_pk}"), 1)
    );
    let mut out = HashMap::new();
    for row in crate::utils::spi::kept_rows(&sql, &[lookup.arg()], lookup.plan())? {
        if let [Some(key), data] = row.as_slice() {
            let data = data
                .as_deref()
                .map_or(Ok(Value::Null), serde_json::from_str)
                .map_err(|e| TViewError::SerializationError {
                    message: format!("tv_{entity}: unreadable data for {key}: {e}"),
                })?;
            out.insert(key.clone(), data);
        }
    }
    Ok(out)
}

fn build_report(
    changes: &[NetChange],
    overflow: &BTreeSet<String>,
    max_entities: usize,
    include_data: bool,
    wanted: Option<&BTreeSet<String>>,
) -> TViewResult<Value> {
    let entities: BTreeSet<&str> = changes
        .iter()
        .map(|c| c.entity.as_str())
        .chain(overflow.iter().map(String::as_str))
        .collect();
    let info = entity_info(&entities)?;
    let typename = |entity: &str| {
        info.get(entity).map_or_else(
            || crate::catalog::typename(entity, None),
            |i| i.typename.clone(),
        )
    };
    let asked = |entity: &str| wanted.is_none_or(|w| w.contains(&typename(entity)));

    // Rows of the types not asked for are left out before the limit applies.
    let omitted: BTreeSet<String> = changes
        .iter()
        .map(|c| c.entity.as_str())
        .chain(overflow.iter().map(String::as_str))
        .filter(|entity| !asked(entity))
        .map(typename)
        .collect();

    // Flush order is not deterministic across runs, so report in a fixed order:
    // entity, then key (numerically when the keys are integers).
    let mut changes: Vec<NetChange> = changes
        .iter()
        .filter(|c| asked(&c.entity))
        .cloned()
        .collect();
    changes.sort_by(|a, b| {
        a.entity
            .cmp(&b.entity)
            .then_with(|| match (a.pk.parse::<i64>(), b.pk.parse::<i64>()) {
                (Ok(x), Ok(y)) => x.cmp(&y),
                _ => a.pk.cmp(&b.pk),
            })
    });
    let (reported, left_out) = changes.split_at(changes.len().min(max_entities));

    // The fresh documents, one read per entity, only when asked for: the ids
    // come from the journal.
    let mut rows: HashMap<(String, String), Value> = HashMap::new();
    if include_data {
        let mut pks_by_entity: HashMap<&str, Vec<String>> = HashMap::new();
        for c in reported.iter().filter(|c| c.change != Change::Deleted) {
            pks_by_entity
                .entry(&c.entity)
                .or_default()
                .push(c.pk.clone());
        }
        for (entity, pks) in pks_by_entity {
            if let Some(i) = info.get(entity) {
                for (pk, data) in current_data(entity, &i.table, &pks)? {
                    rows.insert((entity.to_string(), pk), data);
                }
            }
        }
    }

    let mut updated = Vec::new();
    let mut deleted = Vec::new();
    for c in reported {
        let mut item = Map::new();
        item.insert("__typename".into(), Value::String(typename(&c.entity)));
        item.insert(
            "id".into(),
            c.id.as_ref()
                .map_or(Value::Null, crate::catalog::RowId::value),
        );
        if c.change == Change::Deleted {
            deleted.push(Value::Object(item));
            continue;
        }
        let op = if c.change == Change::Inserted {
            "CREATED"
        } else {
            "UPDATED"
        };
        item.insert("operation".into(), Value::String(op.into()));
        if include_data {
            // A row written then removed by a later, unjournaled path is skipped.
            let Some(data) = rows.remove(&(c.entity.clone(), c.pk.clone())) else {
                continue;
            };
            item.insert("data".into(), data);
        }
        updated.push(Value::Object(item));
    }

    let invalidated: BTreeSet<String> = left_out
        .iter()
        .map(|c| c.entity.as_str())
        .chain(overflow.iter().map(String::as_str).filter(|e| asked(e)))
        .map(typename)
        .collect();
    let mut report = json!({
        "updated": updated,
        "deleted": deleted,
        "truncated": !invalidated.is_empty(),
        "invalidated_types": invalidated.into_iter().collect::<Vec<_>>(),
    });
    if wanted.is_some() {
        report["omitted_types"] = json!(omitted.into_iter().collect::<Vec<_>>());
    }
    Ok(report)
}