kanade-backend 0.31.1

//! v0.31 / #41: change-only history projection.
//!
//! Pairs with the `explode` module — those derived tables are the
//! "what's installed now" snapshot, this module logs "what changed
//! since the previous scan". The projector diff fires BEFORE the
//! explode DELETE-then-INSERT replace so the comparison sees prior
//! state.
//!
//! Volume control: only writes when something actually changed.
//! No-op scans (most of them, once the fleet stabilises) produce
//! zero rows. Combined with the 90-day default retention sweeper
//! (see `cleanup` module), volume tracks fleet churn rather than
//! scan cadence.
//!
//! Identity: arrays are matched element-to-element using the
//! `spec.primary_key` tuple. `identity_json` on each change event
//! serialises just those columns so queries like "every PC ever
//! running Chrome" can filter without per-manifest schema.

use anyhow::Result;
use kanade_shared::manifest::ExplodeSpec;
use serde_json::Value as JsonValue;
use sqlx::{Sqlite, Transaction};
use std::collections::HashMap;

/// One change event ready to insert into `inventory_history`.
/// Constructed by [`diff_explode_rows`] and consumed by
/// [`write_events`].
#[derive(Debug, Clone, PartialEq)]
pub struct HistoryEvent {
    pub change_kind: &'static str, // "added" | "removed" | "changed"
    pub identity_json: String,
    pub before_json: Option<String>,
    pub after_json: Option<String>,
}

/// Read the prior rows for `(pc_id, job_id)` from the derived
/// explode table and diff against the incoming `arr` (the payload's
/// `spec.field` array). Returns the list of change events to
/// persist. The diff is keyed on `spec.primary_key` — elements that
/// match a prior row are checked for column differences, unmatched
/// prior rows are `removed`, unmatched new elements are `added`.
pub async fn diff_explode_rows(
    tx: &mut Transaction<'_, Sqlite>,
    spec: &ExplodeSpec,
    pc_id: &str,
    job_id: &str,
    arr: &[JsonValue],
) -> Result<Vec<HistoryEvent>> {
    // Pull every prior row for this (pc_id, job_id) — we read into
    // a HashMap keyed by the primary-key tuple so the per-element
    // lookup is O(1). The select column list is built from the
    // spec so the test-time table shape stays in sync with the
    // manifest's declaration.
    let select_cols: Vec<String> = spec
        .columns
        .iter()
        .map(|c| format!("\"{}\"", c.field))
        .collect();
    let prior_sql = format!(
        "SELECT {} FROM \"{}\" WHERE pc_id = ? AND job_id = ?",
        select_cols.join(", "),
        spec.table,
    );
    let prior_rows: Vec<sqlx::sqlite::SqliteRow> = sqlx::query(&prior_sql)
        .bind(pc_id)
        .bind(job_id)
        .fetch_all(&mut **tx)
        .await?;

    // Convert each row to a JSON object so subsequent comparisons
    // are uniform with the incoming `arr` elements.
    let mut prior_by_key: HashMap<String, JsonValue> = HashMap::with_capacity(prior_rows.len());
    for row in &prior_rows {
        let obj = row_to_json(row, spec);
        let key = identity_string(&obj, &spec.primary_key);
        prior_by_key.insert(key, obj);
    }

    // Walk incoming elements: each one either matches a prior row
    // (compare for `changed`) or is new (`added`). Track seen keys
    // so the remaining prior_by_key entries become `removed`.
    let mut events = Vec::new();
    let mut seen_keys: std::collections::HashSet<String> =
        std::collections::HashSet::with_capacity(arr.len());

    for element in arr {
        let key = identity_string(element, &spec.primary_key);
        // CodeRabbit #86 fix: skip duplicate identities in the same
        // incoming payload. Pre-fix every duplicate generated an
        // additional `added` / `changed` event, but the explode
        // INSERT step drops the duplicate on PK conflict — so the
        // history could record an event that doesn't match the
        // final snapshot state. `HashSet::insert` returns false
        // when the value was already present, so the early-continue
        // collapses N duplicates to one history event.
        if !seen_keys.insert(key.clone()) {
            continue;
        }

        let identity_json = identity_json_for(element, &spec.primary_key);
        match prior_by_key.get(&key) {
            None => events.push(HistoryEvent {
                change_kind: "added",
                identity_json,
                before_json: None,
                after_json: Some(serde_json::to_string(element)?),
            }),
            Some(prior) if rows_differ(prior, element, spec) => events.push(HistoryEvent {
                change_kind: "changed",
                identity_json,
                before_json: Some(serde_json::to_string(prior)?),
                after_json: Some(serde_json::to_string(element)?),
            }),
            Some(_) => { /* identical — no event */ }
        }
    }

    for (key, prior) in &prior_by_key {
        if seen_keys.contains(key) {
            continue;
        }
        events.push(HistoryEvent {
            change_kind: "removed",
            identity_json: identity_json_for(prior, &spec.primary_key),
            before_json: Some(serde_json::to_string(prior)?),
            after_json: None,
        });
    }

    Ok(events)
}

/// Persist a batch of change events to `inventory_history` inside
/// the caller's transaction. Empty `events` is a no-op (typical
/// case for fleet-stable scans). `field_path` is the manifest
/// field name (`apps`, `disks`) so the timeline can be filtered
/// per field.
pub async fn write_events(
    tx: &mut Transaction<'_, Sqlite>,
    pc_id: &str,
    job_id: &str,
    field_path: &str,
    events: &[HistoryEvent],
) -> Result<()> {
    if events.is_empty() {
        return Ok(());
    }
    // Gemini #86 fix: single multi-VALUES insert via QueryBuilder
    // instead of N round-trips. SQLite likes batch INSERTs and the
    // transaction overhead per statement is non-trivial when a PC
    // turns over a hundred installed apps. Still inside the caller's
    // transaction for atomicity vs the DELETE-INSERT replace.
    let mut qb = sqlx::QueryBuilder::<Sqlite>::new(
        "INSERT INTO inventory_history (
             pc_id, job_id, field_path, identity_json,
             change_kind, before_json, after_json
         ) ",
    );
    qb.push_values(events, |mut b, ev| {
        b.push_bind(pc_id)
            .push_bind(job_id)
            .push_bind(field_path)
            .push_bind(&ev.identity_json)
            .push_bind(ev.change_kind)
            .push_bind(&ev.before_json)
            .push_bind(&ev.after_json);
    });
    qb.build().execute(&mut **tx).await?;
    Ok(())
}

/// Stable string key built from the primary_key tuple's values
/// inside one element. Used to match prior rows ↔ new elements
/// inside [`diff_explode_rows`]; sorted via the explicit pk order
/// so the same physical row produces the same string regardless
/// of JSON object iteration order.
fn identity_string(obj: &JsonValue, primary_key: &[String]) -> String {
    let mut parts = Vec::with_capacity(primary_key.len());
    for k in primary_key {
        let v = obj.get(k).cloned().unwrap_or(JsonValue::Null);
        parts.push(format!("{k}={v}"));
    }
    parts.join("|")
}

/// Serialise just the primary_key fields as a JSON object — the
/// payload for `inventory_history.identity_json`.
fn identity_json_for(obj: &JsonValue, primary_key: &[String]) -> String {
    let mut map = serde_json::Map::new();
    for k in primary_key {
        let v = obj.get(k).cloned().unwrap_or(JsonValue::Null);
        map.insert(k.clone(), v);
    }
    serde_json::Value::Object(map).to_string()
}

/// Compare two element-shaped JSON objects across every declared
/// column. Returns true if any non-key column differs. We don't
/// dive into nested arrays inside an element — that level of
/// granularity is rare and would need its own identity contract.
fn rows_differ(a: &JsonValue, b: &JsonValue, spec: &ExplodeSpec) -> bool {
    for col in &spec.columns {
        let av = a.get(&col.field);
        let bv = b.get(&col.field);
        if av != bv {
            return true;
        }
    }
    false
}

/// Convert a SqliteRow back to a serde JSON object using the
/// spec's column list + declared kinds. Used to bring prior rows
/// into a comparable shape for [`rows_differ`].
///
/// Gemini #86 fix: decode errors (schema drift, manifest typo
/// renaming a column, etc.) now warn-log instead of being silently
/// swallowed via `.ok().flatten()`. The value still falls back to
/// JsonValue::Null so the diff can continue — but the warning
/// gives operators a breadcrumb when "every scan produces a
/// changed event" turns out to be a column-decode bug, not real
/// data churn.
fn row_to_json(row: &sqlx::sqlite::SqliteRow, spec: &ExplodeSpec) -> JsonValue {
    use sqlx::Row;
    let mut map = serde_json::Map::new();
    for col in &spec.columns {
        let v: JsonValue = match col.kind.as_deref() {
            Some("integer") => match row.try_get::<Option<i64>, _>(col.field.as_str()) {
                Ok(Some(i)) => JsonValue::Number(i.into()),
                Ok(None) => JsonValue::Null,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        column = %col.field,
                        kind = "integer",
                        "history diff: row decode failed (treating as NULL — diff may generate a false 'changed' event)",
                    );
                    JsonValue::Null
                }
            },
            Some("real") => match row.try_get::<Option<f64>, _>(col.field.as_str()) {
                Ok(Some(f)) => serde_json::Number::from_f64(f)
                    .map(JsonValue::Number)
                    .unwrap_or(JsonValue::Null),
                Ok(None) => JsonValue::Null,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        column = %col.field,
                        kind = "real",
                        "history diff: row decode failed (treating as NULL)",
                    );
                    JsonValue::Null
                }
            },
            _ => match row.try_get::<Option<String>, _>(col.field.as_str()) {
                Ok(Some(s)) => JsonValue::String(s),
                Ok(None) => JsonValue::Null,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        column = %col.field,
                        kind = "text",
                        "history diff: row decode failed (treating as NULL)",
                    );
                    JsonValue::Null
                }
            },
        };
        map.insert(col.field.clone(), v);
    }
    JsonValue::Object(map)
}

#[cfg(test)]
mod tests {
    use super::*;
    use kanade_shared::manifest::{ExplodeColumn, ExplodeSpec};
    use sqlx::SqlitePool;
    use sqlx::sqlite::SqlitePoolOptions;

    fn sample_apps_spec() -> ExplodeSpec {
        ExplodeSpec {
            field: "apps".into(),
            table: "inventory_sw_apps".into(),
            primary_key: vec!["name".into(), "source".into()],
            columns: vec![
                ExplodeColumn {
                    field: "source".into(),
                    kind: Some("text".into()),
                    index: false,
                },
                ExplodeColumn {
                    field: "name".into(),
                    kind: None,
                    index: true,
                },
                ExplodeColumn {
                    field: "version".into(),
                    kind: None,
                    index: false,
                },
            ],
            track_history: true,
        }
    }

    async fn fresh_pool_with_table() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        crate::projector::explode::ensure_table(&pool, &sample_apps_spec())
            .await
            .unwrap();
        pool
    }

    async fn seed_row(pool: &SqlitePool, name: &str, source: &str, version: &str) {
        sqlx::query(
            "INSERT INTO inventory_sw_apps
                (pc_id, job_id, source, name, version)
             VALUES ('pc-1', 'inventory-sw', ?, ?, ?)",
        )
        .bind(source)
        .bind(name)
        .bind(version)
        .execute(pool)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn first_ever_scan_produces_added_events_for_every_element() {
        let pool = fresh_pool_with_table().await;
        let spec = sample_apps_spec();
        let arr = vec![
            serde_json::json!({"name": "Chrome", "source": "msi", "version": "120"}),
            serde_json::json!({"name": "Firefox", "source": "wow6432", "version": "122"}),
        ];
        let mut tx = pool.begin().await.unwrap();
        let events = diff_explode_rows(&mut tx, &spec, "pc-1", "inventory-sw", &arr)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        assert_eq!(events.len(), 2);
        assert!(events.iter().all(|e| e.change_kind == "added"));
        assert!(events.iter().all(|e| e.before_json.is_none()));
        assert!(events.iter().all(|e| e.after_json.is_some()));
    }

    #[tokio::test]
    async fn stable_scan_produces_no_events() {
        let pool = fresh_pool_with_table().await;
        seed_row(&pool, "Chrome", "msi", "120").await;
        let spec = sample_apps_spec();
        let arr = vec![serde_json::json!({"name":"Chrome","source":"msi","version":"120"})];
        let mut tx = pool.begin().await.unwrap();
        let events = diff_explode_rows(&mut tx, &spec, "pc-1", "inventory-sw", &arr)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        assert!(
            events.is_empty(),
            "identical scans must produce zero events"
        );
    }

    #[tokio::test]
    async fn version_change_produces_changed_event_with_before_after() {
        let pool = fresh_pool_with_table().await;
        seed_row(&pool, "Chrome", "msi", "120").await;
        let spec = sample_apps_spec();
        let arr = vec![serde_json::json!({"name":"Chrome","source":"msi","version":"121"})];
        let mut tx = pool.begin().await.unwrap();
        let events = diff_explode_rows(&mut tx, &spec, "pc-1", "inventory-sw", &arr)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        assert_eq!(events.len(), 1);
        let ev = &events[0];
        assert_eq!(ev.change_kind, "changed");
        assert!(
            ev.before_json
                .as_ref()
                .unwrap()
                .contains("\"version\":\"120\"")
        );
        assert!(
            ev.after_json
                .as_ref()
                .unwrap()
                .contains("\"version\":\"121\"")
        );
        // identity_json carries the key tuple so cross-PC search can
        // filter on "this exact app" regardless of version.
        assert!(ev.identity_json.contains("\"name\":\"Chrome\""));
        assert!(ev.identity_json.contains("\"source\":\"msi\""));
    }

    #[tokio::test]
    async fn uninstall_produces_removed_event() {
        let pool = fresh_pool_with_table().await;
        seed_row(&pool, "Chrome", "msi", "120").await;
        seed_row(&pool, "Firefox", "wow6432", "122").await;
        let spec = sample_apps_spec();
        // New scan only has Firefox — Chrome was uninstalled.
        let arr = vec![serde_json::json!({"name":"Firefox","source":"wow6432","version":"122"})];
        let mut tx = pool.begin().await.unwrap();
        let events = diff_explode_rows(&mut tx, &spec, "pc-1", "inventory-sw", &arr)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        assert_eq!(events.len(), 1);
        let ev = &events[0];
        assert_eq!(ev.change_kind, "removed");
        assert!(ev.after_json.is_none());
        assert!(
            ev.before_json
                .as_ref()
                .unwrap()
                .contains("\"name\":\"Chrome\"")
        );
    }

    #[tokio::test]
    async fn mixed_diff_emits_all_three_kinds() {
        let pool = fresh_pool_with_table().await;
        seed_row(&pool, "Chrome", "msi", "120").await; // will change
        seed_row(&pool, "Firefox", "wow6432", "122").await; // will be removed
        let spec = sample_apps_spec();
        let arr = vec![
            serde_json::json!({"name":"Chrome","source":"msi","version":"121"}), // changed
            serde_json::json!({"name":"Edge","source":"appx","version":"122"}),  // added
        ];
        let mut tx = pool.begin().await.unwrap();
        let events = diff_explode_rows(&mut tx, &spec, "pc-1", "inventory-sw", &arr)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        assert_eq!(events.len(), 3);
        let kinds: std::collections::HashSet<_> = events.iter().map(|e| e.change_kind).collect();
        assert!(kinds.contains("added"));
        assert!(kinds.contains("removed"));
        assert!(kinds.contains("changed"));
    }

    #[tokio::test]
    async fn write_events_persists_to_inventory_history() {
        let pool = fresh_pool_with_table().await;
        let events = vec![HistoryEvent {
            change_kind: "added",
            identity_json: r#"{"name":"Chrome"}"#.into(),
            before_json: None,
            after_json: Some(r#"{"name":"Chrome","version":"120"}"#.into()),
        }];
        let mut tx = pool.begin().await.unwrap();
        write_events(&mut tx, "pc-1", "inventory-sw", "apps", &events)
            .await
            .unwrap();
        tx.commit().await.unwrap();
        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM inventory_history")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 1);
        let row: (
            String,
            String,
            String,
            String,
            Option<String>,
            Option<String>,
        ) = sqlx::query_as(
            "SELECT pc_id, job_id, field_path, change_kind, before_json, after_json \
             FROM inventory_history",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(row.0, "pc-1");
        assert_eq!(row.1, "inventory-sw");
        assert_eq!(row.2, "apps");
        assert_eq!(row.3, "added");
        assert_eq!(row.4, None);
        assert!(row.5.unwrap().contains("Chrome"));
    }
}