armdb 0.14.3

//! Metadata-only format steps (`FormatMigration::unchanged`).

use armour_core::Typ;
use armour_typ::{SchemaTyp, widens};
use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
use serde_json::value::RawValue;
use std::sync::Arc;

use crate::codec::SchemaEncoding;
use crate::error::{DbError, DbResult};
use crate::format_migration::plan::MigrationPlan;

/// Schema of a collection's stored rows, persisted in `db.info`.
#[derive(Debug, Clone)]
pub struct StoredSchema {
    pub encoding: SchemaEncoding,
    typ: Arc<SchemaJson>,
}

#[derive(Debug)]
struct SchemaJson {
    text: String,
    // db.info stores the Typ JSON as a JSON string. Cache that escaped string
    // once so every metadata update can copy it without scanning it again.
    escaped: Box<RawValue>,
}

impl PartialEq for StoredSchema {
    fn eq(&self, other: &Self) -> bool {
        self.encoding == other.encoding && self.typ.text == other.typ.text
    }
}

impl Eq for StoredSchema {}

impl Serialize for StoredSchema {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut state = serializer.serialize_struct("StoredSchema", 2)?;
        state.serialize_field("encoding", &self.encoding)?;
        state.serialize_field("typ", &self.typ.escaped)?;
        state.end()
    }
}

impl<'de> Deserialize<'de> for StoredSchema {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        struct Stored {
            encoding: SchemaEncoding,
            typ: String,
        }
        let Stored { encoding, typ } = Stored::deserialize(deserializer)?;
        Self::from_json(encoding, typ).map_err(serde::de::Error::custom)
    }
}

impl StoredSchema {
    /// The stored `Typ` JSON. Immutable so metadata snapshots can share its
    /// cached JSON encoding without stale serialization after a field edit.
    pub fn typ(&self) -> &str {
        &self.typ.text
    }

    fn from_json(encoding: SchemaEncoding, text: String) -> serde_json::Result<Self> {
        let escaped = serde_json::value::to_raw_value(&text)?;
        Ok(Self {
            encoding,
            typ: Arc::new(SchemaJson { text, escaped }),
        })
    }

    pub(crate) fn of(encoding: SchemaEncoding, typ: &Typ) -> DbResult<Self> {
        let typ = serde_json::to_string(typ)
            .map_err(|err| DbError::FormatMismatch(format!("schema serialization: {err}")))?;
        Self::from_json(encoding, typ)
            .map_err(|err| DbError::FormatMismatch(format!("schema serialization: {err}")))
    }

    pub(crate) fn parse(&self) -> DbResult<SchemaTyp> {
        serde_json::from_str(self.typ())
            .map_err(|err| DbError::FormatMismatch(format!("stored schema is unreadable: {err}")))
    }
}

/// A transition applied by `FormatMigration::unchanged`: rows stored at
/// `from_version` with fingerprint `from_typ_hash` are byte-identical at
/// `to_version`. Kept forever: pending txlog records and lagging `format.v`
/// markers refer to it after restarts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct UnchangedRecord {
    pub from_version: u16,
    pub to_version: u16,
    pub from_typ_hash: u64,
}

/// Oldest version whose bytes equal `source`'s through recorded transitions.
pub(crate) fn physical_floor(history: &[UnchangedRecord], source: u16) -> u16 {
    let mut floor = source;
    while let Some(record) = history
        .iter()
        .find(|record| record.to_version == floor && record.from_version < floor)
    {
        floor = record.from_version;
    }
    floor
}

/// Prove every unchanged run of `plan` before any durable write.
pub(crate) fn verify_unchanged(
    plan: &MigrationPlan,
    stored: Option<&StoredSchema>,
    encoding: Option<SchemaEncoding>,
    target: &Typ,
) -> DbResult<()> {
    for run in &plan.unchanged_runs {
        if encoding != Some(SchemaEncoding::Rapira) {
            return Err(DbError::FormatUnsupportedCodec {
                collection: plan.collection,
                from: run.from_version,
            });
        }
        let before = match (run.before_typ, run.starts_at_stored) {
            (Some(typ), _) => SchemaTyp::from(typ),
            (None, true) => {
                let stored = stored.ok_or(DbError::FormatSchemaUnknown {
                    collection: plan.collection,
                    version: run.from_version,
                })?;
                if stored.encoding != SchemaEncoding::Rapira {
                    return Err(DbError::FormatUnsupportedCodec {
                        collection: plan.collection,
                        from: run.from_version,
                    });
                }
                stored.parse()?
            }
            (None, false) => {
                return Err(DbError::FormatSchemaUnknown {
                    collection: plan.collection,
                    version: run.from_version,
                });
            }
        };
        let after = SchemaTyp::from(run.after_typ.unwrap_or(target));
        widens(&before, &after).map_err(|err| DbError::FormatIncompatible {
            collection: plan.collection,
            from: run.from_version,
            path: err.path,
            reason: err.reason,
        })?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use armour_core::GetType;

    fn probe_schema() -> StoredSchema {
        StoredSchema::of(
            SchemaEncoding::Rapira,
            &Typ::Struct(armour_typ::StructType {
                name: "MetadataProbe",
                fields: armour_typ::Fields::Unnamed(&[String::TYPE; 64]),
            }),
        )
        .unwrap()
    }

    #[test]
    fn stored_schema_clone_shares_immutable_json_and_preserves_serialization() {
        let schema = probe_schema();
        let cloned = schema.clone();
        assert!(
            std::ptr::eq(schema.typ(), cloned.typ()),
            "metadata snapshots must share immutable schema text"
        );
        #[derive(Serialize)]
        struct Previous<'a> {
            encoding: SchemaEncoding,
            typ: &'a str,
        }
        let previous = Previous {
            encoding: schema.encoding,
            typ: schema.typ(),
        };
        assert_eq!(
            serde_json::to_vec(&schema).unwrap(),
            serde_json::to_vec(&previous).unwrap()
        );
        assert_eq!(
            serde_json::to_vec_pretty(&schema).unwrap(),
            serde_json::to_vec_pretty(&previous).unwrap()
        );
        let decoded: StoredSchema =
            serde_json::from_slice(&serde_json::to_vec_pretty(&schema).unwrap()).unwrap();
        assert_eq!(decoded, schema);
        assert_eq!(decoded.parse().unwrap(), schema.parse().unwrap());
        assert!(Arc::ptr_eq(&schema.typ, &cloned.typ));
        assert!(Arc::ptr_eq(&decoded.typ, &decoded.clone().typ));
    }
}