biggus-dickus/crates/db/src/catalog.rs

//! Catalogue objects (the inventory-minimum core). Writes record audit entries
//! on the caller's connection, so the change and its audit entry commit together.

use domain::{
    AuditAction, AuditActor, CatalogueObject, FieldChange, FieldType, NewAuditEvent, ObjectId,
    ObjectInput, Visibility,
};
use serde_json::{Value, json};
use sqlx::Row;

use crate::{audit, authority, fields, vocab};

/// The entity_type recorded in the audit log for catalogue objects.
const ENTITY_TYPE: &str = "object";

const OBJECT_COLUMNS: &str = "id, object_number, object_name, number_of_objects, \
     brief_description, current_location, current_owner, recorder, recording_date, \
     visibility, fields, created_at, updated_at";

/// Create an object and record a `created` audit entry, both on `conn`
/// (pass a transaction connection `&mut *tx` so they commit atomically).
pub async fn create_object(
    conn: &mut sqlx::PgConnection,
    actor: AuditActor,
    input: &ObjectInput,
) -> Result<ObjectId, sqlx::Error> {
    let id = ObjectId::new();

    sqlx::query(
        "INSERT INTO object \
         (id, object_number, object_name, number_of_objects, brief_description, \
          current_location, current_owner, recorder, recording_date, visibility) \
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
    )
    .bind(id.to_uuid())
    .bind(&input.object_number)
    .bind(&input.object_name)
    .bind(input.number_of_objects)
    .bind(input.brief_description.as_deref())
    .bind(input.current_location.as_deref())
    .bind(input.current_owner.as_deref())
    .bind(input.recorder.as_deref())
    .bind(input.recording_date)
    .bind(input.visibility.as_str())
    .execute(&mut *conn)
    .await?;

    let changes = creation_changes(input);

    audit::record(
        &mut *conn,
        &NewAuditEvent {
            actor,
            action: AuditAction::Created,
            entity_type: ENTITY_TYPE.to_owned(),
            entity_id: id.to_uuid(),
            changes,
        },
    )
    .await?;

    Ok(id)
}

/// Fetch one object by id.
pub async fn object_by_id<'e, E>(
    executor: E,
    id: ObjectId,
) -> Result<Option<CatalogueObject>, sqlx::Error>
where
    E: sqlx::PgExecutor<'e>,
{
    let sql = format!("SELECT {OBJECT_COLUMNS} FROM object WHERE id = $1");

    let row = sqlx::query(&sql)
        .bind(id.to_uuid())
        .fetch_optional(executor)
        .await?;

    row.map(map_object).transpose()
}

/// List all objects, ordered by object number.
pub async fn list_objects<'e, E>(executor: E) -> Result<Vec<CatalogueObject>, sqlx::Error>
where
    E: sqlx::PgExecutor<'e>,
{
    // TODO: add LIMIT/keyset pagination before exposing this via the API.
    let sql = format!("SELECT {OBJECT_COLUMNS} FROM object ORDER BY object_number");

    let rows = sqlx::query(&sql).fetch_all(executor).await?;

    rows.into_iter().map(map_object).collect()
}

fn map_object(row: sqlx::postgres::PgRow) -> Result<CatalogueObject, sqlx::Error> {
    let visibility_str: String = row.try_get("visibility")?;
    let visibility = Visibility::from_db(&visibility_str).ok_or_else(|| {
        sqlx::Error::Decode(format!("unknown visibility: {visibility_str}").into())
    })?;

    Ok(CatalogueObject {
        id: ObjectId::from_uuid(row.try_get("id")?),
        object_number: row.try_get("object_number")?,
        object_name: row.try_get("object_name")?,
        number_of_objects: row.try_get("number_of_objects")?,
        brief_description: row.try_get("brief_description")?,
        current_location: row.try_get("current_location")?,
        current_owner: row.try_get("current_owner")?,
        recorder: row.try_get("recorder")?,
        recording_date: row.try_get("recording_date")?,
        visibility,
        fields: row.try_get("fields")?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
    })
}

/// The mutable fields as `(name, value)` pairs, for building audit diffs.
/// `None` means the field is unset (NULL).
fn field_values(input: &ObjectInput) -> Vec<(&'static str, Option<Value>)> {
    vec![
        ("object_number", Some(json!(input.object_number))),
        ("object_name", Some(json!(input.object_name))),
        ("number_of_objects", Some(json!(input.number_of_objects))),
        (
            "brief_description",
            input.brief_description.as_ref().map(|v| json!(v)),
        ),
        (
            "current_location",
            input.current_location.as_ref().map(|v| json!(v)),
        ),
        (
            "current_owner",
            input.current_owner.as_ref().map(|v| json!(v)),
        ),
        ("recorder", input.recorder.as_ref().map(|v| json!(v))),
        ("recording_date", input.recording_date.map(|d| json!(d))),
        ("visibility", Some(json!(input.visibility.as_str()))),
    ]
}

/// Audit changes for a newly created object: every set field as an `after` value.
/// Unset (`None`) optional fields are omitted — absence is conveyed by their not
/// appearing, consistent with `FieldChange`'s `None`-means-no-value convention.
fn creation_changes(input: &ObjectInput) -> Vec<FieldChange> {
    field_values(input)
        .into_iter()
        .filter_map(|(field, after)| {
            after.map(|a| FieldChange {
                field: field.to_owned(),
                before: None,
                after: Some(a),
            })
        })
        .collect()
}

/// Audit changes between two field sets: only the fields whose value changed.
fn update_changes(old: &ObjectInput, new: &ObjectInput) -> Vec<FieldChange> {
    field_values(old)
        .into_iter()
        .zip(field_values(new))
        .filter_map(|((field, before), (_, after))| {
            if before != after {
                Some(FieldChange {
                    field: field.to_owned(),
                    before,
                    after,
                })
            } else {
                None
            }
        })
        .collect()
}

/// Update an object and record an `updated` audit entry with field-level diffs,
/// both on `conn`. Returns `false` if the object does not exist. A no-op update
/// (no fields changed) records no audit entry.
pub async fn update_object(
    conn: &mut sqlx::PgConnection,
    actor: AuditActor,
    id: ObjectId,
    input: &ObjectInput,
) -> Result<bool, sqlx::Error> {
    let Some(old) = object_by_id(&mut *conn, id).await? else {
        return Ok(false);
    };

    let changes = update_changes(&old.to_input(), input);
    if changes.is_empty() {
        // No-op: don't touch updated_at or the audit log.
        return Ok(true);
    }

    sqlx::query(
        "UPDATE object SET \
         object_number = $2, object_name = $3, number_of_objects = $4, \
         brief_description = $5, current_location = $6, current_owner = $7, \
         recorder = $8, recording_date = $9, visibility = $10, updated_at = now() \
         WHERE id = $1",
    )
    .bind(id.to_uuid())
    .bind(&input.object_number)
    .bind(&input.object_name)
    .bind(input.number_of_objects)
    .bind(input.brief_description.as_deref())
    .bind(input.current_location.as_deref())
    .bind(input.current_owner.as_deref())
    .bind(input.recorder.as_deref())
    .bind(input.recording_date)
    .bind(input.visibility.as_str())
    .execute(&mut *conn)
    .await?;

    audit::record(
        &mut *conn,
        &NewAuditEvent {
            actor,
            action: AuditAction::Updated,
            entity_type: ENTITY_TYPE.to_owned(),
            entity_id: id.to_uuid(),
            changes,
        },
    )
    .await?;

    Ok(true)
}

/// Delete an object and record a `deleted` audit entry, both on `conn`.
/// Returns `false` if the object did not exist.
pub async fn delete_object(
    conn: &mut sqlx::PgConnection,
    actor: AuditActor,
    id: ObjectId,
) -> Result<bool, sqlx::Error> {
    let result = sqlx::query("DELETE FROM object WHERE id = $1")
        .bind(id.to_uuid())
        .execute(&mut *conn)
        .await?;

    if result.rows_affected() == 0 {
        return Ok(false);
    }

    audit::record(
        &mut *conn,
        &NewAuditEvent {
            actor,
            action: AuditAction::Deleted,
            entity_type: ENTITY_TYPE.to_owned(),
            entity_id: id.to_uuid(),
            changes: Vec::new(),
        },
    )
    .await?;

    Ok(true)
}

/// Why setting flexible field values failed.
#[derive(Debug, thiserror::Error)]
pub enum FieldError {
    #[error("object not found")]
    ObjectNotFound,
    #[error("unknown field: {0}")]
    UnknownField(String),
    #[error("field `{field}` expects a {expected} value")]
    TypeMismatch {
        field: String,
        expected: &'static str,
    },
    #[error("field `{field}`: value does not resolve to an existing {kind}")]
    Unresolved { field: String, kind: &'static str },
    #[error(transparent)]
    Db(#[from] sqlx::Error),
}

/// Replace an object's flexible field values, validating each against the registry
/// (type + term/authority resolution), and audit the per-field diff — all on `conn`.
/// A no-op (identical to the current values) writes nothing and records no audit.
pub async fn set_object_fields(
    conn: &mut sqlx::PgConnection,
    actor: AuditActor,
    object_id: ObjectId,
    values: &serde_json::Map<String, Value>,
) -> Result<(), FieldError> {
    let Some(old) = object_by_id(&mut *conn, object_id).await? else {
        return Err(FieldError::ObjectNotFound);
    };

    for (key, value) in values {
        validate_field(&mut *conn, key, value).await?;
    }

    let new_fields = Value::Object(values.clone());
    let changes = field_map_changes(&old.fields, &new_fields);

    if changes.is_empty() {
        return Ok(());
    }

    sqlx::query("UPDATE object SET fields = $2, updated_at = now() WHERE id = $1")
        .bind(object_id.to_uuid())
        .bind(&new_fields)
        .execute(&mut *conn)
        .await?;

    audit::record(
        &mut *conn,
        &NewAuditEvent {
            actor,
            action: AuditAction::Updated,
            entity_type: ENTITY_TYPE.to_owned(),
            entity_id: object_id.to_uuid(),
            changes,
        },
    )
    .await?;

    Ok(())
}

async fn validate_field(
    conn: &mut sqlx::PgConnection,
    key: &str,
    value: &Value,
) -> Result<(), FieldError> {
    let def = fields::field_definition_by_key(&mut *conn, key)
        .await?
        .ok_or_else(|| FieldError::UnknownField(key.to_owned()))?;

    match def.field_type {
        FieldType::Text => require(value.is_string(), key, "text")?,
        FieldType::LocalizedText => require(
            value
                .as_object()
                .is_some_and(|o| o.values().all(Value::is_string)),
            key,
            "localized-text object {lang: string}",
        )?,
        FieldType::Integer => require(value.is_i64(), key, "integer")?,
        FieldType::Date => require(value.is_string(), key, "date string")?,
        FieldType::Boolean => require(value.is_boolean(), key, "boolean")?,
        FieldType::Term { vocabulary_id } => {
            let term_id = parse_uuid(value, key, "term id (uuid string)")?;

            if vocab::resolve_term(
                &mut *conn,
                vocabulary_id,
                domain::TermId::from_uuid(term_id),
            )
            .await?
            .is_none()
            {
                return Err(FieldError::Unresolved {
                    field: key.to_owned(),
                    kind: "term",
                });
            }
        }
        FieldType::Authority { kind } => {
            let authority_id = parse_uuid(value, key, "authority id (uuid string)")?;

            match authority::resolve_authority(
                &mut *conn,
                domain::AuthorityId::from_uuid(authority_id),
            )
            .await?
            {
                Some(ref_) if kind.is_none_or(|k| ref_.kind() == k) => {}
                _ => {
                    return Err(FieldError::Unresolved {
                        field: key.to_owned(),
                        kind: "authority",
                    });
                }
            }
        }
    }

    Ok(())
}

fn require(ok: bool, field: &str, expected: &'static str) -> Result<(), FieldError> {
    if ok {
        Ok(())
    } else {
        Err(FieldError::TypeMismatch {
            field: field.to_owned(),
            expected,
        })
    }
}

fn parse_uuid(
    value: &Value,
    field: &str,
    expected: &'static str,
) -> Result<uuid::Uuid, FieldError> {
    value
        .as_str()
        .and_then(|s| s.parse::<uuid::Uuid>().ok())
        .ok_or_else(|| FieldError::TypeMismatch {
            field: field.to_owned(),
            expected,
        })
}

/// Per-key diff between two flexible-field maps. `before`/`after` are `None` when
/// the key is absent on that side (so adds and removes are captured).
fn field_map_changes(old: &Value, new: &Value) -> Vec<FieldChange> {
    let empty = serde_json::Map::new();
    let old_map = old.as_object().unwrap_or(&empty);
    let new_map = new.as_object().unwrap_or(&empty);

    let keys: std::collections::BTreeSet<&String> = old_map.keys().chain(new_map.keys()).collect();

    keys.into_iter()
        .filter_map(|key| {
            let before = old_map.get(key).cloned();
            let after = new_map.get(key).cloned();

            if before != after {
                Some(FieldChange {
                    field: key.clone(),
                    before,
                    after,
                })
            } else {
                None
            }
        })
        .collect()
}