path: root/crdt/src/lib.rs

mod vector_clock;

use std::{collections::BTreeSet, fmt::Display};

use num_rational::Ratio;
use uuid::Uuid;

use crate::vector_clock::VectorClock;

use thiserror::Error;

#[derive(Error, Debug)]
pub enum Error {
    #[error("object with ID {0} not found")]
    NotFound(DerivedId),
}

#[derive(Default, Clone)]
pub struct State {
    ops: Vec<Op>,
}

impl State {
    pub fn from_ops(actor_id: &Uuid, data: &[OpData]) -> Self {
        let mut clock = VectorClock::new();

        let ops = data
            .iter()
            .cloned()
            .map(|data| {
                clock = clock.inc(&actor_id);

                Op {
                    id: Uuid::now_v7(),
                    clock: clock.clone(),
                    data,
                }
            })
            .collect();

        State { ops }
    }

    pub fn append_op(&mut self, actor_id: &Uuid, data: OpData) {
        // Increment the last clock for the provided actor
        let clock = self
            .ops
            .last()
            .map(|Op { clock, .. }| clock.inc(actor_id))
            // For an empty document, initialize a new clock
            .unwrap_or_else(|| VectorClock::default().inc(actor_id));

        self.ops.push(Op {
            id: Uuid::now_v7(),
            clock,
            data,
        });
    }

    pub fn merge(&mut self, other: &State) {
        let op_ids: BTreeSet<Uuid> = self.ops.iter().map(|op| op.id).collect();

        for op in &other.ops {
            if !op_ids.contains(&op.id) {
                self.ops.push(op.clone());
            }
        }

        self.ops.sort_by(|op1, op2| {
            op1.clock
                .partial_cmp(&op2.clock)
                // Tie-breaker: yse op ID
                .unwrap_or_else(|| op1.id.cmp(&op2.id))
        });
    }

    pub fn realize(&self) -> Result<Doc, Error> {
        let mut doc = Doc::default();

        for op in &self.ops {
            op.apply(&mut doc)?;
        }

        Ok(doc)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Op {
    id: Uuid,
    clock: VectorClock,
    data: OpData,
}

impl Op {
    fn apply(&self, doc: &mut Doc) -> Result<(), Error> {
        match &self.data {
            OpData::CreateGrid {
                rows,
                base_cells_per_row,
            } => {
                let duration: Ratio<u32> = Ratio::new(1, *base_cells_per_row as u32);

                let rows = (0..*rows)
                    .map(|row_idx| {
                        let cells = (0..*base_cells_per_row)
                            .map(|cell_idx| {
                                Entry::active(Cell {
                                    id: self.id.derive_id("cell", cell_idx),
                                    duration,
                                })
                            })
                            .collect();

                        Entry::active(Row {
                            id: self.id.derive_id("row", row_idx),
                            cells,
                        })
                    })
                    .collect();

                doc.grids.push(Entry::active(Grid {
                    id: self.id.derive_id("grid", 0),
                    rows,
                }));
            }

            OpData::ChangeSubdivisions {
                grid_id,
                row_id,
                start_cell_id,
                end_cell_id,
                subdivisions,
            } => {
                let grid = doc
                    .grids
                    .iter_mut()
                    .find(|entry| entry.is_active_and(|g| g.id == *grid_id))
                    .ok_or(Error::NotFound(grid_id.clone()))?;

                let row = grid
                    .value_mut()
                    .rows
                    .iter_mut()
                    .find(|entry| entry.is_active_and(|r| r.id == *row_id))
                    .ok_or(Error::NotFound(row_id.clone()))?;

                let start_cell_idx = row
                    .value()
                    .cells
                    .iter()
                    .position(|entry| entry.is_active_and(|c| c.id == *start_cell_id))
                    .ok_or(Error::NotFound(start_cell_id.clone()))?;

                let end_cell_idx = row
                    .value()
                    .cells
                    .iter()
                    .position(|entry| entry.is_active_and(|c| c.id == *end_cell_id))
                    .ok_or(Error::NotFound(end_cell_id.clone()))?;

                let (i, j) = if start_cell_idx <= end_cell_idx {
                    (start_cell_idx, end_cell_idx)
                } else {
                    (end_cell_idx, start_cell_idx)
                };

                let span_duration: Ratio<u32> = row.value().cells[i..j + 1]
                    .iter()
                    .map(|cell| cell.value().duration)
                    .sum();

                let duration: Ratio<u32> = span_duration / *subdivisions as u32;

                row.value_mut().cells.splice(
                    i..(j + 1),
                    (0..*subdivisions).map(|subdivision_idx| {
                        Entry::active(Cell {
                            id: self.id.derive_id("cell", subdivision_idx),
                            duration,
                        })
                    }),
                );
            }
        }

        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpData {
    CreateGrid {
        rows: usize,
        base_cells_per_row: usize,
    },

    ChangeSubdivisions {
        grid_id: DerivedId,
        row_id: DerivedId,
        start_cell_id: DerivedId,
        end_cell_id: DerivedId,
        subdivisions: usize,
    },
}

#[derive(Default, Debug)]
pub struct Doc {
    grids: Vec<Entry<Grid>>,
}

#[derive(Debug)]
pub enum Entry<T> {
    Active { value: T },
    Deleted { value: T, deleted_by: Uuid },
}

impl<T> Entry<T> {
    pub fn active(value: T) -> Self {
        Self::Active { value }
    }

    pub fn is_active_and(&self, f: impl FnOnce(&T) -> bool) -> bool {
        match self {
            Self::Active { value } => f(value),
            _ => false,
        }
    }

    pub fn value(&self) -> &T {
        match self {
            Self::Active { value } => value,
            Self::Deleted { value, .. } => value,
        }
    }

    pub fn value_mut(&mut self) -> &mut T {
        match self {
            Self::Active { value } => value,
            Self::Deleted { value, .. } => value,
        }
    }
}

#[derive(Debug)]
pub struct Grid {
    id: DerivedId,
    rows: Vec<Entry<Row>>,
}

#[derive(Debug)]
pub struct Row {
    id: DerivedId,
    cells: Vec<Entry<Cell>>,
}

#[derive(Debug)]
pub struct Cell {
    id: DerivedId,
    duration: Ratio<u32>,
}

#[derive(PartialEq, Eq, Debug, Clone, Copy)]
pub struct DerivedId {
    // TODO These IDs can be interned on the Doc
    id: Uuid,
    tag: &'static str,
    index: usize,
}

impl Display for DerivedId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}:{}:{}", self.id, self.tag, self.index)
    }
}

trait DerivableId {
    fn derive_id(&self, tag: &'static str, index: usize) -> DerivedId;
}

impl DerivableId for Uuid {
    fn derive_id(&self, tag: &'static str, index: usize) -> DerivedId {
        DerivedId {
            id: self.clone(),
            tag,
            index,
        }
    }
}

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

    #[test]
    fn merge() {
        let actor1 = Uuid::now_v7();
        let actor2 = Uuid::now_v7();

        let mut doc1 = State::from_ops(
            &actor1,
            &[OpData::CreateGrid {
                rows: 4,
                base_cells_per_row: 16,
            }],
        );

        let mut doc2 = State::from_ops(
            &actor2,
            &[OpData::CreateGrid {
                rows: 4,
                base_cells_per_row: 16,
            }],
        );

        doc1.merge(&doc2);

        assert_eq!(doc1.ops.len(), 2);
        assert_eq!(doc1.ops.last().unwrap(), doc2.ops.last().unwrap());

        doc2.merge(&doc1);

        assert_eq!(doc2.ops.len(), 2);
    }

    #[test]
    fn concurrent_ops() {
        let actor1 = Uuid::now_v7();
        let actor2 = Uuid::now_v7();

        let mut doc1 = State::from_ops(
            &actor1,
            &[OpData::CreateGrid {
                rows: 4,
                base_cells_per_row: 16,
            }],
        );

        let mut doc2 = doc1.clone();

        {
            let realized = doc1.realize().unwrap();

            doc1.append_op(
                &actor1,
                OpData::ChangeSubdivisions {
                    grid_id: realized.grids[0].value().id,
                    row_id: realized.grids[0].value().rows[0].value().id,
                    start_cell_id: realized.grids[0].value().rows[0].value().cells[0]
                        .value()
                        .id,
                    end_cell_id: realized.grids[0].value().rows[0].value().cells[3]
                        .value()
                        .id,
                    subdivisions: 3,
                },
            );

            doc2.append_op(
                &actor2,
                OpData::ChangeSubdivisions {
                    grid_id: realized.grids[0].value().id,
                    row_id: realized.grids[0].value().rows[0].value().id,
                    start_cell_id: realized.grids[0].value().rows[0].value().cells[0]
                        .value()
                        .id,
                    end_cell_id: realized.grids[0].value().rows[0].value().cells[3]
                        .value()
                        .id,
                    subdivisions: 3,
                },
            );
        }

        assert_eq!(
            doc1.ops
                .last()
                .unwrap()
                .clock
                .partial_cmp(&doc2.ops.last().unwrap().clock),
            None
        );

        doc1.merge(&doc2);

        assert_eq!(doc1.ops.len(), 3);

        let realized = doc1.realize().unwrap();

        let grid = &realized.grids[0].value();
        let row = &grid.rows[0].value();

        assert_eq!(
            row.cells
                .iter()
                .map(|cell| cell.value().duration)
                .sum::<Ratio<u32>>(),
            Ratio::ONE
        );
    }

    #[test]
    fn realize_doc() {
        let actor_id = Uuid::now_v7();

        let mut doc = State::default();

        doc.append_op(
            &actor_id,
            OpData::CreateGrid {
                rows: 4,
                base_cells_per_row: 16,
            },
        );

        {
            let realized = doc.realize().unwrap();

            assert_eq!(realized.grids.len(), 1);

            let grid = realized.grids.first().unwrap().value();

            assert_eq!(grid.rows.len(), 4);

            let row = grid.rows.first().unwrap().value();

            assert_eq!(row.cells.len(), 16);

            assert_eq!(
                row.cells
                    .iter()
                    .map(|cell| cell.value().duration)
                    .sum::<Ratio<u32>>(),
                Ratio::ONE
            );

            doc.append_op(
                &actor_id,
                OpData::ChangeSubdivisions {
                    grid_id: grid.id.clone(),
                    row_id: row.id.clone(),
                    start_cell_id: row.cells[0].value().id.clone(),
                    end_cell_id: row.cells[3].value().id.clone(),
                    subdivisions: 3,
                },
            );
        }

        {
            let realized = doc.realize().unwrap();

            assert_eq!(realized.grids[0].value().rows[0].value().cells.len(), 15);
            assert_eq!(realized.grids[0].value().rows[1].value().cells.len(), 16);

            let grid = &realized.grids[0].value();
            let row = &grid.rows[0].value();

            assert_eq!(
                row.cells
                    .iter()
                    .map(|cell| cell.value().duration)
                    .sum::<Ratio<u32>>(),
                Ratio::ONE
            );

            doc.append_op(
                &actor_id,
                OpData::ChangeSubdivisions {
                    grid_id: grid.id.clone(),
                    row_id: row.id.clone(),
                    start_cell_id: row.cells[0].value().id.clone(),
                    end_cell_id: row.cells.last().unwrap().value().id.clone(),
                    subdivisions: 12,
                },
            );
        }

        {
            let realized = doc.realize().unwrap();

            let grid = &realized.grids[0].value();
            let row = &grid.rows[0].value();

            assert_eq!(row.cells.len(), 12);
            assert_eq!(realized.grids[0].value().rows[1].value().cells.len(), 16);

            assert_eq!(
                row.cells
                    .iter()
                    .map(|cell| cell.value().duration)
                    .sum::<Ratio<u32>>(),
                Ratio::ONE
            );
        }
    }
}