second_brain_sync/
export.rs

1use std::io::Write;
2
3use anyhow::Result;
4use chrono::{DateTime, Utc};
5use serde::{Deserialize, Serialize};
6
7use second_brain_core::kuzu_store::KuzuStore;
8use second_brain_core::schema::{SyncNodeType, SyncOp};
9
10#[derive(Debug, Serialize, Deserialize)]
11pub struct SyncRecord {
12    pub local_seq: u64,
13    pub origin_machine_id: String,
14    pub origin_seq: u64,
15    pub op: SyncOp,
16    pub node_type: SyncNodeType,
17    pub node_id: String,
18    pub timestamp: DateTime<Utc>,
19    pub data: Option<serde_json::Value>,
20}
21
22pub fn export_changes<W: Write>(store: &KuzuStore, after_seq: u64, writer: &mut W) -> Result<u64> {
23    let entries = store.sync_log_since(after_seq)?;
24    let mut max_seq = after_seq;
25
26    let mut records: Vec<SyncRecord> = entries
27        .into_iter()
28        .map(|e| {
29            let data = e.data.as_ref().and_then(|d| serde_json::from_str(d).ok());
30            SyncRecord {
31                local_seq: e.local_seq as u64,
32                origin_machine_id: e.origin_machine_id,
33                origin_seq: e.origin_seq as u64,
34                op: e.op,
35                node_type: e.node_type,
36                node_id: e.node_id,
37                timestamp: e.timestamp,
38                data,
39            }
40        })
41        .collect();
42
43    sort_for_import(&mut records);
44
45    for record in &records {
46        serde_json::to_writer(&mut *writer, record)?;
47        writer.write_all(b"\n")?;
48        if record.local_seq > max_seq {
49            max_seq = record.local_seq;
50        }
51    }
52
53    Ok(max_seq)
54}
55
56pub fn export_to_stdout(store: &KuzuStore, after_seq: u64) -> Result<u64> {
57    let stdout = std::io::stdout();
58    let mut handle = stdout.lock();
59    export_changes(store, after_seq, &mut handle)
60}
61
62pub fn sort_for_import(records: &mut [SyncRecord]) {
63    records.sort_by(|a, b| {
64        let type_order = |r: &SyncRecord| -> u8 {
65            match (&r.op, &r.node_type) {
66                (SyncOp::Delete, SyncNodeType::Relation) => 0,
67                (SyncOp::Delete, SyncNodeType::Memory) => 1,
68                (SyncOp::Delete, SyncNodeType::Entity) => 2,
69                (SyncOp::Delete, SyncNodeType::Conversation) => 3,
70                (SyncOp::Create, SyncNodeType::Conversation) => 4,
71                (SyncOp::Create, SyncNodeType::Entity) => 5,
72                (SyncOp::Create, SyncNodeType::Memory) => 6,
73                (SyncOp::Create, SyncNodeType::Relation) => 7,
74                (SyncOp::Update, SyncNodeType::Conversation) => 8,
75                (SyncOp::Update, SyncNodeType::Entity) => 9,
76                (SyncOp::Update, SyncNodeType::Memory) => 10,
77                (SyncOp::Update, SyncNodeType::Relation) => 11,
78            }
79        };
80        type_order(a)
81            .cmp(&type_order(b))
82            .then(a.local_seq.cmp(&b.local_seq))
83    });
84}
second_brain_sync/export.rs

second_brain_sync/
export.rs
