sqlite-diff-rs 0.9.0

//! Maxwell message parsing and conversion to `SQLite` changeset operations.
//!
//! [Maxwell](https://maxwells-daemon.io/) is a `MySQL` CDC application that
//! reads binlogs and emits row-level changes as JSON. This module deserializes
//! those events and converts them into changeset operations compatible with
//! this crate's builders.
//!
//! Maxwell events carry no trigger-origin marker, so converted ops default
//! to `indirect = false`. Override via the [`Indirect`](crate::Indirect) trait
//! if you know out-of-band that the event was trigger-induced.
//!
//! # Example
//!
//! ```
//! use sqlite_diff_rs::maxwell::{parse, Message, OpType};
//!
//! let json = r#"{"database":"mydb","table":"users","type":"insert","ts":1477053217,"data":{"id":1,"name":"Alice"}}"#;
//! let message = parse(json).unwrap();
//!
//! assert_eq!(message.op_type, OpType::Insert);
//! assert_eq!(message.table, "users");
//! ```

use alloc::collections::BTreeMap;
use alloc::string::String;
use alloc::vec::Vec;
use serde::{Deserialize, Serialize};

/// Maxwell operation type.
///
/// Represents the type of database operation captured by Maxwell.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum OpType {
    /// INSERT operation.
    Insert,
    /// UPDATE operation.
    Update,
    /// DELETE operation.
    Delete,
}

/// Maxwell CDC message.
///
/// Represents a single row-level change captured from `MySQL` binlog.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Message {
    /// Database name.
    pub database: String,
    /// Table name.
    pub table: String,
    /// Operation type (insert, update, delete).
    #[serde(rename = "type")]
    pub op_type: OpType,
    /// Unix timestamp (seconds) when the change occurred.
    #[serde(default)]
    pub ts: Option<i64>,
    /// Transaction ID.
    #[serde(default)]
    pub xid: Option<i64>,
    /// Whether this is the final row in the transaction.
    #[serde(default)]
    pub commit: Option<bool>,
    /// Binlog position (e.g., "master.000006:800911").
    #[serde(default)]
    pub position: Option<String>,
    /// `MySQL` server ID.
    #[serde(default)]
    pub server_id: Option<i64>,
    /// `MySQL` thread ID.
    #[serde(default)]
    pub thread_id: Option<i64>,
    /// Primary key values.
    #[serde(default)]
    pub primary_key: Option<Vec<serde_json::Value>>,
    /// Primary key column names.
    #[serde(default)]
    pub primary_key_columns: Option<Vec<String>>,
    /// Current row data (new values for insert/update, deleted values for delete).
    pub data: BTreeMap<String, serde_json::Value>,
    /// Previous values for changed columns (update only).
    #[serde(default)]
    pub old: Option<BTreeMap<String, serde_json::Value>>,
    /// Per-column MySQL type names emitted when the Maxwell daemon
    /// runs with `--include_types`. Absent otherwise.
    #[serde(default)]
    pub columns_types: Option<BTreeMap<String, String>>,
}

/// Parse a Maxwell message from JSON.
///
/// # Errors
///
/// Returns a [`serde_json::Error`] if the JSON is malformed.
///
/// # Example
///
/// ```
/// use sqlite_diff_rs::maxwell::{parse, OpType};
///
/// let json = r#"{"database":"test","table":"users","type":"insert","data":{"id":1}}"#;
/// let msg = parse(json).unwrap();
/// assert_eq!(msg.op_type, OpType::Insert);
/// ```
pub fn parse(json: &str) -> Result<Message, serde_json::Error> {
    serde_json::from_str(json)
}

pub use crate::wire::ConversionError;

use crate::ChangesetFormat;
use crate::builders::{ChangeDelete, Insert, PatchDelete, Update};
use crate::schema::NamedColumns;

use crate::wire::{
    Sealed, WireColumnItem, WireSource, WireType, build_changeset_delete, build_insert,
    build_patch_delete, build_patchset_update, resolve_table,
};

use crate::builders::{DiffOps, DiffSetBuilder, PatchsetFormat};
use crate::wire::WireAdapter;
use core::fmt::Debug;
use core::hash::Hash;

/// Marker type for the `maxwell` source.
#[derive(Debug, Clone, Copy, Default)]
pub struct Maxwell;

impl Sealed for Maxwell {}

impl WireSource for Maxwell {
    type Payload<'a> = MaxwellColumn<'a>;

    fn wire_type(payload: &Self::Payload<'_>) -> WireType {
        payload.wire_type
    }

    fn column_name<'a>(payload: &'a Self::Payload<'_>) -> &'a str {
        payload.column_name
    }
}

/// Per-column payload for the `maxwell` source.
///
/// The schema supplies the semantic [`WireType`] per column, independent
/// of the MySQL type name Maxwell may emit.
#[derive(Debug, Clone, Copy)]
pub struct MaxwellColumn<'a> {
    /// Column name.
    pub column_name: &'a str,
    /// Semantic column type driving decoder dispatch.
    pub wire_type: WireType,
    /// Column value as a JSON value.
    pub value: &'a serde_json::Value,
}

/// One Maxwell column adapted to the shared [`WireColumnItem`] contract.
struct MaxwellItem<'a> {
    name: &'a str,
    value: &'a serde_json::Value,
}

impl WireColumnItem<Maxwell> for MaxwellItem<'_> {
    fn name(&self) -> &str {
        self.name
    }

    fn payload(&self, wire_type: WireType) -> MaxwellColumn<'_> {
        MaxwellColumn {
            column_name: self.name,
            wire_type,
            value: self.value,
        }
    }
}

impl MaxwellColumn<'_> {
    /// Ergonomic helper for calling a specific [`Decoder`](crate::wire::Decoder) on this
    /// payload without fully-qualified syntax. Fixes the `Src` generic
    /// to [`Maxwell`] so the compiler can pick the impl.
    ///
    /// # Errors
    ///
    /// Propagates the decoder's [`DecodeError`](crate::wire::DecodeError).
    pub fn decoded_by<D, S, B>(
        self,
        decoder: &D,
    ) -> Result<crate::encoding::Value<S, B>, crate::wire::DecodeError>
    where
        D: crate::wire::Decoder<Maxwell, S, B>,
    {
        decoder.decode(self)
    }
}

use crate::wire::{Digestable, WireColumnTypes, WireSchema};

impl<T, S, B> Digestable<ChangesetFormat, T, S, B> for Message
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + Debug + Hash + Eq + AsRef<str> + Default,
    B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default,
{
    type Src = Maxwell;
    type Error = ConversionError;

    fn digest_into<Sch, A>(
        &self,
        builder: DiffSetBuilder<ChangesetFormat, T, S, B>,
        schema: &Sch,
        adapter: &A,
    ) -> Result<DiffSetBuilder<ChangesetFormat, T, S, B>, ConversionError>
    where
        Sch: WireSchema<Table = T>,
        A: WireAdapter<Maxwell, S, B>,
    {
        let table = resolve_table(schema, self.table.as_str())?;
        match self.op_type {
            OpType::Insert => {
                let insert = build_insert_from_maxwell(&self.data, table, adapter)?;
                Ok(DiffOps::insert(builder, insert))
            }
            OpType::Update => {
                let update = build_changeset_update_from_maxwell(
                    &self.data,
                    self.old.as_ref(),
                    table,
                    adapter,
                )?;
                Ok(DiffOps::update(builder, update))
            }
            OpType::Delete => {
                let delete = build_changeset_delete_from_maxwell(&self.data, table, adapter)?;
                Ok(DiffOps::delete(builder, delete))
            }
        }
    }
}

impl<T, S, B> Digestable<PatchsetFormat, T, S, B> for Message
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + Debug + Hash + Eq + AsRef<str> + Default,
    B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default,
{
    type Src = Maxwell;
    type Error = ConversionError;

    fn digest_into<Sch, A>(
        &self,
        builder: DiffSetBuilder<PatchsetFormat, T, S, B>,
        schema: &Sch,
        adapter: &A,
    ) -> Result<DiffSetBuilder<PatchsetFormat, T, S, B>, ConversionError>
    where
        Sch: WireSchema<Table = T>,
        A: WireAdapter<Maxwell, S, B>,
    {
        let table = resolve_table(schema, self.table.as_str())?;
        match self.op_type {
            OpType::Insert => {
                let insert = build_insert_from_maxwell(&self.data, table, adapter)?;
                Ok(DiffOps::insert(builder, insert))
            }
            OpType::Update => {
                let update = build_patchset_update_from_maxwell(&self.data, table, adapter)?;
                Ok(DiffOps::update(builder, update))
            }
            OpType::Delete => {
                let delete = build_patch_delete_from_maxwell(&self.data, table, adapter)?;
                Ok(DiffOps::delete(builder, delete))
            }
        }
    }
}

fn build_insert_from_maxwell<T, S, B, A>(
    data: &BTreeMap<String, serde_json::Value>,
    table: &T,
    adapter: &A,
) -> Result<Insert<T, S, B>, ConversionError>
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + AsRef<str>,
    B: Clone + AsRef<[u8]>,
    A: WireAdapter<Maxwell, S, B>,
{
    build_insert(
        data.iter().map(|(name, value)| MaxwellItem {
            name: name.as_str(),
            value,
        }),
        table,
        adapter,
    )
}

fn build_changeset_update_from_maxwell<T, S, B, A>(
    data: &BTreeMap<String, serde_json::Value>,
    old: Option<&BTreeMap<String, serde_json::Value>>,
    table: &T,
    adapter: &A,
) -> Result<Update<T, ChangesetFormat, S, B>, ConversionError>
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + Debug + AsRef<str>,
    B: Clone + Debug + AsRef<[u8]>,
    A: WireAdapter<Maxwell, S, B>,
{
    let mut update: Update<T, ChangesetFormat, S, B> = Update::from(table.clone());
    for (name, new_value) in data {
        let col_idx = table
            .column_index(name)
            .ok_or_else(|| ConversionError::ColumnNotFound(name.clone()))?;
        let wire_type = table.column_type(col_idx);

        let new_payload = MaxwellColumn {
            column_name: name.as_str(),
            wire_type,
            value: new_value,
        };
        let new = adapter.decode(new_payload)?;

        // A column present in `old` changed, so pair its old and new values. A
        // column absent from `old` did not change, so its old value equals the
        // new value. That keeps an unchanged primary key available for the
        // WHERE clause and unchanged columns out of the SET.
        let old = if let Some(old_value) = old.and_then(|old_map| old_map.get(name)) {
            let old_payload = MaxwellColumn {
                column_name: name.as_str(),
                wire_type,
                value: old_value,
            };
            adapter.decode(old_payload)?
        } else {
            new.clone()
        };
        update = update
            .set(col_idx, old, new)
            .map_err(|_| ConversionError::ColumnNotFound(name.clone()))?;
    }
    Ok(update)
}

fn build_patchset_update_from_maxwell<T, S, B, A>(
    data: &BTreeMap<String, serde_json::Value>,
    table: &T,
    adapter: &A,
) -> Result<Update<T, PatchsetFormat, S, B>, ConversionError>
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + AsRef<str>,
    B: Clone + AsRef<[u8]>,
    A: WireAdapter<Maxwell, S, B>,
{
    build_patchset_update(
        data.iter().map(|(name, value)| MaxwellItem {
            name: name.as_str(),
            value,
        }),
        table,
        adapter,
    )
}

fn build_changeset_delete_from_maxwell<T, S, B, A>(
    data: &BTreeMap<String, serde_json::Value>,
    table: &T,
    adapter: &A,
) -> Result<ChangeDelete<T, S, B>, ConversionError>
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + Default + AsRef<str>,
    B: Clone + Default + AsRef<[u8]>,
    A: WireAdapter<Maxwell, S, B>,
{
    build_changeset_delete(
        data.iter().map(|(name, value)| MaxwellItem {
            name: name.as_str(),
            value,
        }),
        table,
        adapter,
    )
}

fn build_patch_delete_from_maxwell<T, S, B, A>(
    data: &BTreeMap<String, serde_json::Value>,
    table: &T,
    adapter: &A,
) -> Result<PatchDelete<T, S, B>, ConversionError>
where
    T: NamedColumns + WireColumnTypes,
    S: Clone + AsRef<str>,
    B: Clone + AsRef<[u8]>,
    A: WireAdapter<Maxwell, S, B>,
{
    build_patch_delete(
        data.iter().map(|(name, value)| MaxwellItem {
            name: name.as_str(),
            value,
        }),
        table,
        adapter,
    )
}

// ============================================================================
// Arbitrary implementations for testing
// ============================================================================

#[cfg(feature = "testing")]
mod arbitrary_impl {
    use super::{Message, OpType};
    use alloc::collections::BTreeMap;
    use alloc::string::ToString;
    use arbitrary::{Arbitrary, Unstructured};

    impl<'a> Arbitrary<'a> for OpType {
        fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
            Ok(*u.choose(&[Self::Insert, Self::Update, Self::Delete])?)
        }
    }

    impl<'a> Arbitrary<'a> for Message {
        fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
            let op_type = OpType::arbitrary(u)?;

            // Generate 1-5 columns of data
            let num_cols: usize = u.int_in_range(1..=5)?;
            let mut data = BTreeMap::new();
            for i in 0..num_cols {
                let col_name = alloc::format!("col{i}");
                let value: i64 = u.arbitrary()?;
                data.insert(col_name, serde_json::Value::Number(value.into()));
            }

            // For updates, generate old values for some columns
            let old = if op_type == OpType::Update {
                let mut old_data = BTreeMap::new();
                let num_changed: usize = u.int_in_range(1..=num_cols)?;
                for i in 0..num_changed {
                    let col_name = alloc::format!("col{i}");
                    let value: i64 = u.arbitrary()?;
                    old_data.insert(col_name, serde_json::Value::Number(value.into()));
                }
                Some(old_data)
            } else {
                None
            };

            Ok(Self {
                database: "testdb".to_string(),
                table: "testtable".to_string(),
                op_type,
                ts: u.arbitrary()?,
                xid: u.arbitrary()?,
                commit: u.arbitrary()?,
                position: None,
                server_id: u.arbitrary()?,
                thread_id: u.arbitrary()?,
                primary_key: None,
                primary_key_columns: None,
                data,
                old,
                columns_types: None,
            })
        }
    }
}