//! Deterministic snapshot codec fixtures retained for future secure sync work. //! //! SITE-01 exposes no production transport, trigger, credential, response //! handler, or store-mutation path. These pure codecs remain testable so later //! SITE-11/SITE-12 work can replace the legacy protocol under explicit gates. //! The current merge model has no tombstones and is not release-safe. use nigig_site::store::SiteStore; pub const SYNC_PROTOCOL: &str = "nigig-site-sync/1"; /// Stay under the server's 64KB message cap with headroom for the envelope. pub const MAX_CHUNK_BYTES: usize = 60 * 1024; /// Merge `remote` into `local` in place. Returns counts added/updated. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct MergeStats { pub sites_added: usize, pub sites_updated: usize, pub reports_added: usize, pub reports_updated: usize, pub tasks_added: usize, pub tables_added: usize, pub materials_added: usize, pub suppliers_added: usize, pub meetings_added: usize, pub reminders_added: usize, pub contacts_added: usize, pub threads_merged: usize, } pub fn merge_remote(local: &mut SiteStore, remote: SiteStore) -> MergeStats { let mut stats = MergeStats::default(); for site in remote.sites { match local.sites.iter_mut().find(|s| s.id == site.id) { Some(existing) => { if site.updated_at > existing.updated_at { *existing = site; stats.sites_updated += 1; } } None => { local.sites.push(site); stats.sites_added += 1; } } } for rep in remote.reports { match local.reports.iter_mut().find(|r| r.id == rep.id) { Some(existing) => { if rep.updated_at > existing.updated_at { *existing = rep; stats.reports_updated += 1; } } None => { local.reports.push(rep); stats.reports_added += 1; } } } for task in remote.tasks { if !local.tasks.iter().any(|t| t.id == task.id) { local.tasks.push(task); stats.tasks_added += 1; } } for table in remote.workers { if !local .workers .iter() .any(|t| t.site_id == table.site_id && t.date == table.date) { local.workers.push(table); stats.tables_added += 1; } } for sched in remote.procurement { match local .procurement .iter_mut() .find(|p| p.site_id == sched.site_id) { Some(existing) => { for line in sched.lines { if !existing.lines.iter().any(|l| l.id == line.id) { existing.lines.push(line); stats.materials_added += 1; } } } None => { stats.materials_added += sched.lines.len(); local.procurement.push(sched); } } } for supplier in remote.suppliers.suppliers { if !local .suppliers .suppliers .iter() .any(|s| s.id == supplier.id) { local.suppliers.suppliers.push(supplier); stats.suppliers_added += 1; } } for meeting in remote.meetings { if !local.meetings.iter().any(|m| m.id == meeting.id) { local.meetings.push(meeting); stats.meetings_added += 1; } } for reminder in remote.reminders { if !local.reminders.iter().any(|r| r.id == reminder.id) { local.reminders.push(reminder); stats.reminders_added += 1; } } for dir in remote.directories { match local .directories .iter_mut() .find(|d| d.site_id == dir.site_id) { Some(existing) => { for contact in dir.contacts { if !existing .contacts .iter() .any(|c| c.user_id == contact.user_id) { existing.contacts.push(contact); stats.contacts_added += 1; } } } None => { stats.contacts_added += dir.contacts.len(); local.directories.push(dir); } } } for (room, lines) in remote.chat_threads { let entry = local.chat_threads.entry(room).or_default(); for line in lines { if !entry.contains(&line) { entry.push(line); stats.threads_merged += 1; } } while entry.len() > 50 { entry.remove(0); } } // Selection follows the remote only when we have nothing selected. if local.selected_site_id.is_none() { local.selected_site_id = remote.selected_site_id; } stats } #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] struct ChunkEnvelope { protocol: String, snapshot: String, index: usize, total: usize, payload: String, } /// Split a store snapshot into message-sized chunks. `snapshot_id` should be /// unique per push (ULID); chunks carry their position for reassembly. pub fn encode_snapshot(store: &SiteStore, snapshot_id: &str) -> Vec { let json = serde_json::to_string(store).unwrap_or_else(|_| "{}".to_string()); let bytes = json.as_bytes(); let total = (bytes.len() + MAX_CHUNK_BYTES - 1) / MAX_CHUNK_BYTES.max(1); let total = total.max(1); bytes .chunks(MAX_CHUNK_BYTES) .enumerate() .map(|(index, chunk)| { serde_json::to_string(&ChunkEnvelope { protocol: SYNC_PROTOCOL.to_string(), snapshot: snapshot_id.to_string(), index, total, payload: String::from_utf8_lossy(chunk).into_owned(), }) .unwrap_or_default() }) .collect() } /// Reassemble chunks (any order, duplicates tolerated) back into a store. /// Returns `None` when the set is incomplete or unparseable — never partial. pub fn decode_snapshot(messages: &[String]) -> Option { let mut envelopes: Vec = messages .iter() .filter_map(|m| serde_json::from_str::(m).ok()) .filter(|e| e.protocol == SYNC_PROTOCOL) .collect(); if envelopes.is_empty() { return None; } // Newest snapshot wins when several are present. envelopes.sort_by(|a, b| a.snapshot.cmp(&b.snapshot)); let snapshot = envelopes.last()?.snapshot.clone(); let mut parts: Vec<&ChunkEnvelope> = envelopes .iter() .filter(|e| e.snapshot == snapshot) .collect(); let total = parts.first()?.total; if total == 0 || parts.len() < total { return None; } parts.sort_by_key(|e| e.index); parts.dedup_by_key(|e| e.index); if parts.len() != total { return None; } let joined: String = parts.iter().map(|e| e.payload.as_str()).collect(); serde_json::from_str(&joined).ok() } /// Pull sync chunks out of raw room timeline bodies (defensive: skips /// anything that is not a sync envelope). pub fn extract_chunks_from_bodies(bodies: &[String]) -> Vec { bodies .iter() .filter(|b| b.contains(SYNC_PROTOCOL)) .cloned() .collect() } #[cfg(test)] mod tests { use super::*; use chrono::{NaiveDate, TimeZone, Utc}; use nigig_site::domain::approvals::ConstructionTask; use nigig_site::domain::daily_report::DailyReport; use nigig_site::domain::site::{Site, SiteNature}; fn site(id: &str, updated_at: chrono::DateTime) -> Site { let mut s = Site::new(format!("{id}-name"), SiteNature::Road, "x"); s.id = id.to_string(); s.updated_at = updated_at; s } #[test] fn merge_unions_and_applies_lww() { let t0 = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); let t1 = Utc.with_ymd_and_hms(2026, 9, 2, 0, 0, 0).unwrap(); let mut local = SiteStore::default(); local.sites.push(site("s1", t0)); let mut remote = SiteStore::default(); let mut newer = site("s1", t1); newer.name = "renamed".into(); remote.sites.push(newer); remote.sites.push(site("s2", t0)); let stats = merge_remote(&mut local, remote); assert_eq!(stats.sites_updated, 1); assert_eq!(stats.sites_added, 1); assert_eq!(local.site("s1").unwrap().name, "renamed"); // Older remote never overwrites newer local. let mut stale = SiteStore::default(); stale.sites.push(site("s1", t0)); let stats = merge_remote(&mut local, stale); assert_eq!(stats.sites_updated, 0); assert_eq!(local.site("s1").unwrap().name, "renamed"); } #[test] fn tasks_merge_without_overwrite_and_reports_lww() { let mut local = SiteStore::default(); local.tasks.push(ConstructionTask::new("s1", "local-task")); let mut remote = SiteStore::default(); remote .tasks .push(ConstructionTask::new("s1", "remote-task")); let date = NaiveDate::from_ymd_opt(2026, 9, 7).unwrap(); let mut old = DailyReport::new("s1", date); old.updated_at = Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap(); local.reports.push(old); let mut new = DailyReport::new("s1", date); new.id = local.reports[0].id.clone(); new.updated_at = Utc.with_ymd_and_hms(2026, 9, 8, 0, 0, 0).unwrap(); remote.reports.push(new); let stats = merge_remote(&mut local, remote); assert_eq!(stats.tasks_added, 1); assert_eq!(stats.reports_updated, 1); assert_eq!(local.tasks.len(), 2); } #[test] fn snapshot_chunks_round_trip() { let mut store = SiteStore::default(); store.sites.push(site("s1", Utc::now())); store.tasks.push(ConstructionTask::new("s1", "t")); let chunks = encode_snapshot(&store, "snap-1"); assert!(!chunks.is_empty()); for c in &chunks { assert!(c.len() <= MAX_CHUNK_BYTES + 256); } // Duplicates + shuffle tolerated. let mut mixed = chunks.clone(); mixed.extend(chunks.clone()); mixed.reverse(); let back = decode_snapshot(&mixed).unwrap(); assert_eq!(back.sites.len(), 1); assert_eq!(back.tasks.len(), 1); } #[test] fn incomplete_snapshot_is_refused_not_partial() { let mut store = SiteStore::default(); // Force multiple chunks with a big description. let mut big = ConstructionTask::new("s1", "big"); big.description = Some("x".repeat(MAX_CHUNK_BYTES * 2)); store.tasks.push(big); let chunks = encode_snapshot(&store, "snap-2"); assert!(chunks.len() > 1); assert!(decode_snapshot(&chunks[1..]).is_none()); assert!(decode_snapshot(&[]).is_none()); assert!(decode_snapshot(&["hello".to_string()]).is_none()); } }