446 lines
16 KiB
Rust
446 lines
16 KiB
Rust
//! Task CRDT operations — create, read, mutate, and merge task documents.
|
||
//!
|
||
//! ## Data Flow
|
||
//!
|
||
//! ```text
|
||
//! Dart → task_crdt_create(id, ns, title) → Vec<u8> (serialized Automerge doc)
|
||
//! Dart → db_save_crdt_doc(id, "task", bytes, ts) (persist to SQLCipher)
|
||
//! Dart → db_load_crdt_doc(id) → Vec<u8> (reload from DB)
|
||
//! Dart → task_crdt_read(bytes) → FfiTaskData (structured snapshot)
|
||
//! Dart → task_crdt_set_title(bytes, title) → Vec<u8>(mutated doc bytes)
|
||
//! ```
|
||
//!
|
||
//! Mutation functions (create, set_*) require an active engine because they
|
||
//! advance the node's Hybrid Logical Clock. Read and merge functions are
|
||
//! stateless.
|
||
|
||
use chronogrid_core::TaskCrdt;
|
||
|
||
use super::state::with_engine_mut;
|
||
use super::types::FfiTaskData;
|
||
|
||
// ── Construction ─────────────────────────────────────────────────────────────
|
||
|
||
/// Creates a new task CRDT document and returns its serialized bytes.
|
||
///
|
||
/// The document is initialized with `status = "backlog"`,
|
||
/// `priority_score = 0.0`, and `is_archived = false`.
|
||
///
|
||
/// **Requires active engine** (for HLC).
|
||
pub fn task_crdt_create(
|
||
id: String,
|
||
namespace_id: String,
|
||
title: String,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::new(&id, &namespace_id, &title, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("task_crdt_create failed: {e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
// ── Read ─────────────────────────────────────────────────────────────────────
|
||
|
||
/// Deserializes a task CRDT document and returns a structured snapshot.
|
||
///
|
||
/// **Stateless** — does not require an active engine.
|
||
pub fn task_crdt_read(doc_bytes: Vec<u8>) -> anyhow::Result<FfiTaskData> {
|
||
let task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("task_crdt_read: failed to load doc: {e}"))?;
|
||
|
||
Ok(FfiTaskData {
|
||
id: task.entity_id()?.unwrap_or_default(),
|
||
namespace_id: task.namespace_id()?.unwrap_or_default(),
|
||
project_id: task.project_id()?,
|
||
title: task.title()?.unwrap_or_default(),
|
||
description_markdown: task.description_markdown()?,
|
||
status: task.status()?.unwrap_or_else(|| "backlog".into()),
|
||
priority_score: task.priority_score()?.unwrap_or(0.0),
|
||
eisenhower_quadrant: task.eisenhower_quadrant()?,
|
||
due_date: task.due_date()?.map(|v| v as i64),
|
||
estimated_duration: task.estimated_duration()?,
|
||
is_archived: task.is_archived()?.unwrap_or(false),
|
||
completed_at: task.completed_at()?.map(|v| v as i64),
|
||
hlc_timestamp: task.hlc_timestamp()?.unwrap_or_default(),
|
||
})
|
||
}
|
||
|
||
// ── Field mutations ──────────────────────────────────────────────────────────
|
||
//
|
||
// Each setter loads the doc from bytes, applies the mutation with an
|
||
// advancing HLC, re-serializes, and returns the updated bytes.
|
||
|
||
/// Sets the task title. **Requires active engine.**
|
||
pub fn task_crdt_set_title(doc_bytes: Vec<u8>, title: String) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_title(&title, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the task status (`backlog`, `in_progress`, `review`, `completed`).
|
||
pub fn task_crdt_set_status(doc_bytes: Vec<u8>, status: String) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_status(&status, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the markdown description.
|
||
pub fn task_crdt_set_description(
|
||
doc_bytes: Vec<u8>,
|
||
description: String,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_description_markdown(&description, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Clears the markdown description.
|
||
pub fn task_crdt_clear_description(doc_bytes: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.clear_description(&mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the computed priority score.
|
||
pub fn task_crdt_set_priority_score(
|
||
doc_bytes: Vec<u8>,
|
||
score: f64,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_priority_score(score, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the Eisenhower quadrant (1–4).
|
||
pub fn task_crdt_set_eisenhower_quadrant(
|
||
doc_bytes: Vec<u8>,
|
||
quadrant: i64,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_eisenhower_quadrant(quadrant, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Clears the Eisenhower quadrant.
|
||
pub fn task_crdt_clear_eisenhower_quadrant(doc_bytes: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.clear_eisenhower_quadrant(&mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the due date (Unix epoch milliseconds).
|
||
pub fn task_crdt_set_due_date(doc_bytes: Vec<u8>, due_ms: i64) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_due_date(due_ms as u64, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Clears the due date.
|
||
pub fn task_crdt_clear_due_date(doc_bytes: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.clear_due_date(&mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the estimated duration in minutes.
|
||
pub fn task_crdt_set_estimated_duration(
|
||
doc_bytes: Vec<u8>,
|
||
minutes: i64,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_estimated_duration(minutes, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the archived flag.
|
||
pub fn task_crdt_set_is_archived(
|
||
doc_bytes: Vec<u8>,
|
||
archived: bool,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_is_archived(archived, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Sets the completion timestamp (Unix epoch milliseconds).
|
||
pub fn task_crdt_set_completed_at(
|
||
doc_bytes: Vec<u8>,
|
||
ts_ms: i64,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_completed_at(ts_ms as u64, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Clears the completion timestamp.
|
||
pub fn task_crdt_clear_completed_at(doc_bytes: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.clear_completed_at(&mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Moves the task to a different project.
|
||
pub fn task_crdt_set_project_id(
|
||
doc_bytes: Vec<u8>,
|
||
project_id: String,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_project_id(&project_id, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Removes the project association.
|
||
pub fn task_crdt_clear_project_id(doc_bytes: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.clear_project_id(&mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
/// Moves the task to a different namespace.
|
||
pub fn task_crdt_set_namespace_id(
|
||
doc_bytes: Vec<u8>,
|
||
namespace_id: String,
|
||
) -> anyhow::Result<Vec<u8>> {
|
||
with_engine_mut(|engine| {
|
||
let mut task = TaskCrdt::from_bytes(&doc_bytes)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
task.set_namespace_id(&namespace_id, &mut engine.hlc)
|
||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||
Ok(task.save())
|
||
})
|
||
}
|
||
|
||
// ── Merge ────────────────────────────────────────────────────────────────────
|
||
|
||
/// Merges two task CRDT documents and returns the combined document bytes.
|
||
///
|
||
/// **Stateless** — does not require an active engine. Automerge resolves
|
||
/// concurrent writes deterministically via its built-in LWW semantics.
|
||
pub fn task_crdt_merge(doc_a: Vec<u8>, doc_b: Vec<u8>) -> anyhow::Result<Vec<u8>> {
|
||
let mut task_a = TaskCrdt::from_bytes(&doc_a)
|
||
.map_err(|e| anyhow::anyhow!("task_crdt_merge: failed to load doc_a: {e}"))?;
|
||
let mut task_b = TaskCrdt::from_bytes(&doc_b)
|
||
.map_err(|e| anyhow::anyhow!("task_crdt_merge: failed to load doc_b: {e}"))?;
|
||
|
||
task_a
|
||
.merge(&mut task_b)
|
||
.map_err(|e| anyhow::anyhow!("task_crdt_merge: merge failed: {e}"))?;
|
||
|
||
Ok(task_a.save())
|
||
}
|
||
|
||
// ── Tests ────────────────────────────────────────────────────────────────────
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::super::engine::{close_engine, init_engine_in_memory};
|
||
use super::*;
|
||
|
||
const TEST_KEY: [u8; 32] = [0xAA; 32];
|
||
const NODE: &str = "test-node";
|
||
|
||
fn setup() {
|
||
let _ = close_engine();
|
||
init_engine_in_memory(TEST_KEY.to_vec(), NODE.into()).unwrap();
|
||
}
|
||
|
||
fn teardown() {
|
||
let _ = close_engine();
|
||
}
|
||
|
||
#[test]
|
||
fn create_and_read_round_trip() {
|
||
setup();
|
||
|
||
let bytes = task_crdt_create("t-001".into(), "ns-1".into(), "My Task".into())
|
||
.unwrap();
|
||
let data = task_crdt_read(bytes).unwrap();
|
||
|
||
assert_eq!(data.id, "t-001");
|
||
assert_eq!(data.namespace_id, "ns-1");
|
||
assert_eq!(data.title, "My Task");
|
||
assert_eq!(data.status, "backlog");
|
||
assert!((data.priority_score - 0.0).abs() < f64::EPSILON);
|
||
assert!(!data.is_archived);
|
||
assert!(!data.hlc_timestamp.is_empty());
|
||
|
||
teardown();
|
||
}
|
||
|
||
#[test]
|
||
fn set_all_fields_and_read() {
|
||
setup();
|
||
|
||
let mut bytes =
|
||
task_crdt_create("t-002".into(), "ns-1".into(), "Initial".into()).unwrap();
|
||
|
||
bytes = task_crdt_set_title(bytes, "Updated".into()).unwrap();
|
||
bytes = task_crdt_set_status(bytes, "in_progress".into()).unwrap();
|
||
bytes = task_crdt_set_description(bytes, "# Desc".into()).unwrap();
|
||
bytes = task_crdt_set_priority_score(bytes, 0.75).unwrap();
|
||
bytes = task_crdt_set_eisenhower_quadrant(bytes, 2).unwrap();
|
||
bytes = task_crdt_set_due_date(bytes, 1_700_000_000_000).unwrap();
|
||
bytes = task_crdt_set_estimated_duration(bytes, 60).unwrap();
|
||
bytes = task_crdt_set_is_archived(bytes, true).unwrap();
|
||
bytes = task_crdt_set_completed_at(bytes, 1_700_100_000_000).unwrap();
|
||
bytes = task_crdt_set_project_id(bytes, "p-001".into()).unwrap();
|
||
bytes = task_crdt_set_namespace_id(bytes, "ns-2".into()).unwrap();
|
||
|
||
let data = task_crdt_read(bytes).unwrap();
|
||
|
||
assert_eq!(data.title, "Updated");
|
||
assert_eq!(data.status, "in_progress");
|
||
assert_eq!(data.description_markdown, Some("# Desc".into()));
|
||
assert!((data.priority_score - 0.75).abs() < f64::EPSILON);
|
||
assert_eq!(data.eisenhower_quadrant, Some(2));
|
||
assert_eq!(data.due_date, Some(1_700_000_000_000));
|
||
assert_eq!(data.estimated_duration, Some(60));
|
||
assert!(data.is_archived);
|
||
assert_eq!(data.completed_at, Some(1_700_100_000_000));
|
||
assert_eq!(data.project_id, Some("p-001".into()));
|
||
assert_eq!(data.namespace_id, "ns-2");
|
||
|
||
teardown();
|
||
}
|
||
|
||
#[test]
|
||
fn clear_optional_fields() {
|
||
setup();
|
||
|
||
let mut bytes =
|
||
task_crdt_create("t-003".into(), "ns-1".into(), "Task".into()).unwrap();
|
||
|
||
bytes = task_crdt_set_description(bytes, "desc".into()).unwrap();
|
||
bytes = task_crdt_clear_description(bytes).unwrap();
|
||
|
||
bytes = task_crdt_set_eisenhower_quadrant(bytes, 3).unwrap();
|
||
bytes = task_crdt_clear_eisenhower_quadrant(bytes).unwrap();
|
||
|
||
bytes = task_crdt_set_due_date(bytes, 123).unwrap();
|
||
bytes = task_crdt_clear_due_date(bytes).unwrap();
|
||
|
||
bytes = task_crdt_set_completed_at(bytes, 456).unwrap();
|
||
bytes = task_crdt_clear_completed_at(bytes).unwrap();
|
||
|
||
bytes = task_crdt_set_project_id(bytes, "p-1".into()).unwrap();
|
||
bytes = task_crdt_clear_project_id(bytes).unwrap();
|
||
|
||
let data = task_crdt_read(bytes).unwrap();
|
||
assert_eq!(data.description_markdown, None);
|
||
assert_eq!(data.eisenhower_quadrant, None);
|
||
assert_eq!(data.due_date, None);
|
||
assert_eq!(data.completed_at, None);
|
||
assert_eq!(data.project_id, None);
|
||
|
||
teardown();
|
||
}
|
||
|
||
#[test]
|
||
fn merge_concurrent_edits() {
|
||
setup();
|
||
|
||
let bytes_a =
|
||
task_crdt_create("t-004".into(), "ns-1".into(), "Original".into()).unwrap();
|
||
|
||
// Simulate a concurrent peer by creating a second copy.
|
||
// (In production, forks would come from network sync.)
|
||
let bytes_b = task_crdt_set_title(bytes_a.clone(), "Peer B Title".into()).unwrap();
|
||
let bytes_a = task_crdt_set_title(bytes_a, "Peer A Title".into()).unwrap();
|
||
|
||
let merged = task_crdt_merge(bytes_a, bytes_b).unwrap();
|
||
let data = task_crdt_read(merged).unwrap();
|
||
|
||
// Automerge picks a deterministic winner.
|
||
assert!(
|
||
data.title == "Peer A Title" || data.title == "Peer B Title",
|
||
"merge must resolve to one of the concurrent writes, got: {}",
|
||
data.title
|
||
);
|
||
|
||
teardown();
|
||
}
|
||
|
||
#[test]
|
||
fn read_is_stateless() {
|
||
// Ensure read works without engine.
|
||
let _ = close_engine();
|
||
|
||
// We need bytes from a previous engine session.
|
||
init_engine_in_memory(TEST_KEY.to_vec(), NODE.into()).unwrap();
|
||
let bytes =
|
||
task_crdt_create("t-005".into(), "ns-1".into(), "Stateless".into()).unwrap();
|
||
close_engine().unwrap();
|
||
|
||
// Read after engine shutdown should still work.
|
||
let data = task_crdt_read(bytes).unwrap();
|
||
assert_eq!(data.title, "Stateless");
|
||
}
|
||
|
||
#[test]
|
||
fn mutation_without_engine_fails() {
|
||
let _ = close_engine();
|
||
let result = task_crdt_create("x".into(), "y".into(), "z".into());
|
||
assert!(result.is_err());
|
||
}
|
||
}
|