nigig-org/crates/apps/nigig-site/tests/support/sync_fixture.rs

//! 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<String> {
    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<SiteStore> {
    let mut envelopes: Vec<ChunkEnvelope> = messages
        .iter()
        .filter_map(|m| serde_json::from_str::<ChunkEnvelope>(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<String> {
    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<Utc>) -> 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());
    }
}