mushroomdb-storage 0.1.2

use crate::types::Value;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum WalRecord {
    InsertNode {
        label: String,
        key: String,
        props: Vec<(String, Value)>,
    },
    InsertEdge {
        edge_type: String,
        src_key: String,
        dst_key: String,
    },
    SetProp {
        key: String,
        field: String,
        value: Value,
    },
    CreateRule {
        def_bytes: Vec<u8>,
    },
    DeleteRule {
        name: String,
    },
    // ── Mutation variants (appended last — bincode is positional) ─────────────
    RemoveProp {
        key: String,
        field: String,
    },
    DeleteEdge {
        edge_type: String,
        src_key: String,
        dst_key: String,
    },
    DeleteNode {
        key: String,
    },
    /// One WAL frame = one atomic batch. Nested `Batch` inside a `Batch` is
    /// invalid: `encode_record` debug-asserts against it, and `decode_all`
    /// treats a frame whose payload deserialises to a nested `Batch` as corrupt
    /// (stops cleanly before that frame, returning the valid prefix).
    Batch(Vec<WalRecord>),
    /// Recompute one rule from scratch (un-trip / repair). Appended after
    /// `Batch`; bincode discriminant is 9. `Batch` stays at 8.
    RebuildRule {
        name: String,
    },
    // ── View variants (appended LAST; bincode discriminant is positional) ─────
    /// Create (or restore) a materialized property view.  Bincoded `ViewDef`
    /// bytes keep core-storage free of core-rules types.  Discriminant 10.
    CreateView {
        def_bytes: Vec<u8>,
    },
    /// Delete a named view and remove its values from all nodes.  Discriminant 11.
    DeleteView {
        name: String,
    },
    // ── Full-text-lite variants (appended after views; discriminants 12, 13) ──
    /// Enable full-text indexing on all nodes of `label` for field `field`.
    /// The index is rebuilt from live data on open; this record only persists
    /// the declaration.  Discriminant 12.
    EnableFulltext {
        label: String,
        field: String,
    },
    /// Disable full-text indexing for `(label, field)` and drop its postings.
    /// Discriminant 13.
    DisableFulltext {
        label: String,
        field: String,
    },
    // ── Dense-id variants (appended after full-text; discriminants 14–17) ──
    /// Insert a node using interned label/field ids. Key remains a string once.
    /// Discriminant 14.
    InsertNodeId {
        label: u32,
        key: String,
        props: Vec<(u32, Value)>,
    },
    /// Set a property using dense node id and interned field id. Discriminant 15.
    SetPropId {
        id: u32,
        field: u32,
        value: Value,
    },
    /// Insert an edge using interned etype and dense node ids. Discriminant 16.
    InsertEdgeId {
        etype: u32,
        src: u32,
        dst: u32,
    },
    /// Bind intern id `id` to `text` so subsequent `*Id` records can replay
    /// without a snapshot intern table. Discriminant 17. Apply is idempotent
    /// when the string is already bound to `id`.
    ///
    /// Discriminant order is append-order, not emit-order: in a WAL stream
    /// `Intern` always *precedes* the `*Id` records that reference it, even
    /// though it carries the highest discriminant of the 14–17 group.
    Intern {
        id: u32,
        text: String,
    },
}

/// Encode a single WAL record as a framed byte sequence: `[len u32][crc u32][payload]`.
///
/// # Panics (debug builds)
/// Panics if `rec` is a `Batch` that contains a nested `Batch` — nested batches
/// are semantically invalid.
pub fn encode_record(rec: &WalRecord) -> Vec<u8> {
    if let WalRecord::Batch(inner) = rec {
        debug_assert!(
            !inner.iter().any(|r| matches!(r, WalRecord::Batch(_))),
            "nested Batch is invalid: a Batch may not contain another Batch"
        );
    }
    let payload = bincode::serialize(rec).expect("walrecord serialize cannot fail");
    let crc = crc32fast::hash(&payload);
    let mut out = Vec::with_capacity(8 + payload.len());
    out.extend((payload.len() as u32).to_le_bytes());
    out.extend(crc.to_le_bytes());
    out.extend(payload);
    out
}

/// Count the number of complete, valid WAL frames in `bytes`.
///
/// Each frame counts as one commit — both `Batch` frames (which represent an
/// atomic multi-op commit) and legacy single-op frames (InsertNode, SetProp,
/// CreateRule, etc.).  The numbering used by `GraphDb::open_at` is 0-based:
/// commit 0 is the first frame, commit N-1 is the last frame in a WAL with N
/// total commits.
///
/// Equivalent to `decode_all(bytes).0.len() as u64`; exposed as a named
/// function so commit-count semantics are pinned independently of the decoder.
pub fn wal_commits(bytes: &[u8]) -> u64 {
    decode_all(bytes).0.len() as u64
}

/// Decode as many complete, valid WAL frames as possible from `bytes`.
///
/// Returns `(records, valid_len)` where `valid_len` is the byte offset of the
/// first frame that was torn, corrupt, or undeserializable — callers can
/// truncate the WAL file to `valid_len` to discard the invalid tail.
///
/// A `Batch` frame whose inner record list contains a nested `Batch` is treated
/// as corrupt: decoding stops before that frame (the frame itself is not pushed).
pub fn decode_all(bytes: &[u8]) -> (Vec<WalRecord>, usize) {
    let mut recs = Vec::new();
    let mut pos = 0usize;
    loop {
        if bytes.len() < pos + 8 {
            return (recs, pos);
        }
        let len = u32::from_le_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
        let crc = u32::from_le_bytes(bytes[pos + 4..pos + 8].try_into().unwrap());
        let start = pos + 8;
        if bytes.len() < start + len {
            return (recs, pos); // torn tail
        }
        let payload = &bytes[start..start + len];
        if crc32fast::hash(payload) != crc {
            return (recs, pos); // corrupt tail
        }
        match bincode::deserialize::<WalRecord>(payload) {
            Ok(WalRecord::Batch(inner)) => {
                // Nested Batch inside a Batch is invalid; treat as corrupt frame.
                if inner.iter().any(|r| matches!(r, WalRecord::Batch(_))) {
                    return (recs, pos);
                }
                recs.push(WalRecord::Batch(inner));
            }
            Ok(r) => recs.push(r),
            Err(_) => return (recs, pos),
        }
        pos = start + len;
    }
}

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

    fn sample() -> Vec<WalRecord> {
        vec![
            WalRecord::InsertNode {
                label: "L".into(),
                key: "k1".into(),
                props: vec![("f".into(), Value::Int(1))],
            },
            WalRecord::InsertEdge {
                edge_type: "E".into(),
                src_key: "k1".into(),
                dst_key: "k2".into(),
            },
        ]
    }

    #[test]
    fn roundtrip_multiple_records() {
        let mut bytes = Vec::new();
        for r in sample() {
            bytes.extend(encode_record(&r));
        }
        let (recs, consumed) = decode_all(&bytes);
        assert_eq!(recs, sample());
        assert_eq!(consumed, bytes.len());
    }

    #[test]
    fn torn_tail_is_dropped_whole() {
        let mut bytes = Vec::new();
        for r in sample() {
            bytes.extend(encode_record(&r));
        }
        let full = bytes.len();
        let first = encode_record(&sample()[0]).len();
        bytes.truncate(full - 3); // tear the second record
        let (recs, consumed) = decode_all(&bytes);
        assert_eq!(recs.len(), 1);
        assert_eq!(consumed, first);
    }

    #[test]
    fn corrupt_crc_stops_replay_at_last_valid() {
        let mut bytes = encode_record(&sample()[0]);
        let n = bytes.len();
        bytes[n - 1] ^= 0xFF; // flip a payload byte
        let (recs, consumed) = decode_all(&bytes);
        assert!(recs.is_empty());
        assert_eq!(consumed, 0);
    }

    #[test]
    fn empty_input_is_fine() {
        let (recs, consumed) = decode_all(&[]);
        assert!(recs.is_empty());
        assert_eq!(consumed, 0);
    }

    // ── Task 2: new variant roundtrips ────────────────────────────────────────

    #[test]
    fn roundtrip_remove_prop() {
        let r = WalRecord::RemoveProp {
            key: "n1".into(),
            field: "age".into(),
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn roundtrip_delete_edge() {
        let r = WalRecord::DeleteEdge {
            edge_type: "KNOWS".into(),
            src_key: "a".into(),
            dst_key: "b".into(),
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn roundtrip_delete_node() {
        let r = WalRecord::DeleteNode { key: "x".into() };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn roundtrip_rebuild_rule() {
        let r = WalRecord::RebuildRule { name: "eq".into() };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn batch_of_three_is_one_frame() {
        let inner = vec![
            WalRecord::DeleteNode { key: "a".into() },
            WalRecord::DeleteNode { key: "b".into() },
            WalRecord::DeleteNode { key: "c".into() },
        ];
        let batch = WalRecord::Batch(inner.clone());
        let frame = encode_record(&batch);

        // Exactly ONE frame: one (u32 len + u32 crc) header at offset 0.
        // Verify by calling decode_all on the raw bytes.
        let (recs, consumed) = decode_all(&frame);
        assert_eq!(consumed, frame.len(), "should consume the whole frame");
        assert_eq!(recs.len(), 1, "one decoded record (the Batch)");
        assert_eq!(recs[0], WalRecord::Batch(inner));
    }

    #[test]
    fn torn_mid_batch_frame_drops_whole_batch() {
        // Two plain records before the batch, then a batch frame that is torn.
        let pre = sample();
        let batch = WalRecord::Batch(vec![
            WalRecord::DeleteNode { key: "a".into() },
            WalRecord::DeleteNode { key: "b".into() },
        ]);
        let mut bytes = Vec::new();
        for r in &pre {
            bytes.extend(encode_record(r));
        }
        let batch_start = bytes.len();
        bytes.extend(encode_record(&batch));

        // Truncate 3 bytes inside the batch frame.
        bytes.truncate(bytes.len() - 3);

        let (recs, consumed) = decode_all(&bytes);
        assert_eq!(recs, pre, "only pre-batch records survive");
        assert_eq!(
            consumed, batch_start,
            "valid_len stops at batch frame start"
        );
    }

    #[test]
    #[cfg(debug_assertions)]
    #[should_panic(expected = "nested Batch")]
    fn nested_batch_encode_panics_in_debug() {
        let inner_batch = WalRecord::Batch(vec![WalRecord::DeleteNode { key: "z".into() }]);
        let outer = WalRecord::Batch(vec![inner_batch]);
        encode_record(&outer); // must debug_assert-panic
    }

    /// Pin the exact on-disk wire format for two variants: one pre-existing
    /// (discriminant 0) and the first new variant added in Plan 4 (discriminant 5).
    ///
    /// **If this test fails you have broken every existing database file.**
    /// WAL variants must ONLY be appended — never reordered or inserted.
    /// The frame layout is `[len: u32 LE][crc32: u32 LE][bincode payload]`.
    /// The discriminant is the first 4 bytes of the payload (u32 LE).
    /// `Batch` stays at discriminant 8; `RebuildRule` is 9.
    #[test]
    fn golden_bytes_pin_wire_format() {
        // ── Variant 0: InsertNode { label: "L", key: "k", props: [] } ──────────
        let insert_node = WalRecord::InsertNode {
            label: "L".into(),
            key: "k".into(),
            props: vec![],
        };
        #[rustfmt::skip]
        let expected_insert_node: &[u8] = &[
            // header: len=30 LE, crc32 LE
            30, 0, 0, 0, 114, 69, 253, 24,
            // payload: discriminant=0 (InsertNode)
            0, 0, 0, 0,
            // label "L": len=1, b'L'
            1, 0, 0, 0, 0, 0, 0, 0, 76,
            // key "k": len=1, b'k'
            1, 0, 0, 0, 0, 0, 0, 0, 107,
            // props: len=0
            0, 0, 0, 0, 0, 0, 0, 0,
        ];
        assert_eq!(
            encode_record(&insert_node),
            expected_insert_node,
            "InsertNode wire format changed — this breaks all existing WAL files"
        );

        // ── Variant 5: RemoveProp { key: "n1", field: "age" } ─────────────────
        // This is the first Plan-4 mutation variant; pins the append boundary.
        let remove_prop = WalRecord::RemoveProp {
            key: "n1".into(),
            field: "age".into(),
        };
        #[rustfmt::skip]
        let expected_remove_prop: &[u8] = &[
            // header: len=25 LE, crc32 LE
            25, 0, 0, 0, 35, 214, 55, 239,
            // payload: discriminant=5 (RemoveProp)
            5, 0, 0, 0,
            // key "n1": len=2, b'n', b'1'
            2, 0, 0, 0, 0, 0, 0, 0, 110, 49,
            // field "age": len=3, b'a', b'g', b'e'
            3, 0, 0, 0, 0, 0, 0, 0, 97, 103, 101,
        ];
        assert_eq!(
            encode_record(&remove_prop),
            expected_remove_prop,
            "RemoveProp wire format changed — this breaks all existing WAL files"
        );

        // ── Variant 8: Batch([DeleteNode { key: "z" }]) ──────────────────────
        // Pins discriminant 8 with an exact-bytes golden (payload hardcoded;
        // CRC derived from that payload so the frame is self-consistent).
        // An accidental insertion of any variant before `Batch` in the enum
        // changes the discriminant bytes, which breaks this assertion
        // immediately — preventing silent corruption of every existing WAL file
        // that contains Batch frames.
        let batch_single = WalRecord::Batch(vec![WalRecord::DeleteNode { key: "z".into() }]);
        // Payload: discriminant 8, vec len 1, DeleteNode discriminant 7, key "z".
        // 4 + 8 + 4 + 8 + 1 = 25 bytes.
        #[rustfmt::skip]
        let batch_payload: &[u8] = &[
            // discriminant=8 (Batch)
            8, 0, 0, 0,
            // inner vec len=1 (bincode u64 LE)
            1, 0, 0, 0, 0, 0, 0, 0,
            // DeleteNode discriminant=7
            7, 0, 0, 0,
            // key "z": len=1 (u64 LE), b'z'=122
            1, 0, 0, 0, 0, 0, 0, 0, 122,
        ];
        let batch_crc = crc32fast::hash(batch_payload);
        let mut expected_batch_frame: Vec<u8> = Vec::with_capacity(8 + batch_payload.len());
        expected_batch_frame.extend((batch_payload.len() as u32).to_le_bytes());
        expected_batch_frame.extend(batch_crc.to_le_bytes());
        expected_batch_frame.extend_from_slice(batch_payload);
        assert_eq!(
            encode_record(&batch_single),
            expected_batch_frame,
            "Batch (discriminant 8) wire format changed — a variant may have \
             been inserted before position 8, breaking all existing WAL Batch frames"
        );

        // ── Variant 9: RebuildRule { name: "eq" } ─────────────────────────────
        // Batch remains discriminant 8; this variant is appended after it.
        let rebuild = WalRecord::RebuildRule { name: "eq".into() };
        #[rustfmt::skip]
        let expected_rebuild: &[u8] = &[
            // header: len=14 LE, crc32 LE
            14, 0, 0, 0, 242, 136, 144, 68,
            // payload: discriminant=9 (RebuildRule)
            9, 0, 0, 0,
            // name "eq": len=2, b'e', b'q'
            2, 0, 0, 0, 0, 0, 0, 0, 101, 113,
        ];
        assert_eq!(
            encode_record(&rebuild),
            expected_rebuild,
            "RebuildRule wire format changed — append-only WAL variants"
        );
    }

    // ── Task 3: view variant roundtrips + wire pins ───────────────────────────

    #[test]
    fn roundtrip_create_view() {
        let r = WalRecord::CreateView {
            def_bytes: vec![1, 2, 3],
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn roundtrip_delete_view() {
        let r = WalRecord::DeleteView {
            name: "my_view".into(),
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    /// Pin discriminants 10 (CreateView) and 11 (DeleteView).
    ///
    /// **If this test fails you have broken every existing database file.**
    /// WAL variants must ONLY be appended — never reordered or inserted.
    #[test]
    fn golden_bytes_pin_view_wire_format() {
        // ── Variant 10: CreateView { def_bytes: [0xDE, 0xAD] } ───────────────
        let create_view = WalRecord::CreateView {
            def_bytes: vec![0xDE, 0xAD],
        };
        let cv_payload = bincode::serialize(&create_view).unwrap();
        // discriminant must be 10 (0x0a 0x00 0x00 0x00 in LE)
        assert_eq!(
            &cv_payload[0..4],
            &[10, 0, 0, 0],
            "CreateView discriminant changed — a variant was inserted before position 10"
        );

        // ── Variant 11: DeleteView { name: "v" } ─────────────────────────────
        let delete_view = WalRecord::DeleteView { name: "v".into() };
        let dv_payload = bincode::serialize(&delete_view).unwrap();
        assert_eq!(
            &dv_payload[0..4],
            &[11, 0, 0, 0],
            "DeleteView discriminant changed — a variant was inserted before position 11"
        );

        // Roundtrip both through encode_record / decode_all.
        let mut buf = encode_record(&create_view);
        buf.extend(encode_record(&delete_view));
        let (recs, consumed) = decode_all(&buf);
        assert_eq!(consumed, buf.len());
        assert_eq!(recs.len(), 2);
        assert_eq!(
            recs[0],
            WalRecord::CreateView {
                def_bytes: vec![0xDE, 0xAD]
            }
        );
        assert_eq!(recs[1], WalRecord::DeleteView { name: "v".into() });
    }

    // ── Task full-text-lite: new variant roundtrips + wire pin ────────────────

    #[test]
    fn roundtrip_enable_fulltext() {
        let r = WalRecord::EnableFulltext {
            label: "Person".into(),
            field: "bio".into(),
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    #[test]
    fn roundtrip_disable_fulltext() {
        let r = WalRecord::DisableFulltext {
            label: "Person".into(),
            field: "bio".into(),
        };
        let bytes = encode_record(&r);
        let (recs, _) = decode_all(&bytes);
        assert_eq!(recs, vec![r]);
    }

    /// Pin discriminants 12 (EnableFulltext) and 13 (DisableFulltext).
    ///
    /// **If this test fails you have broken every existing database file.**
    /// WAL variants must ONLY be appended — never reordered or inserted.
    #[test]
    fn golden_bytes_pin_fulltext_wire_format() {
        // ── Variant 12: EnableFulltext { label: "A", field: "b" } ────────────
        let enable = WalRecord::EnableFulltext {
            label: "A".into(),
            field: "b".into(),
        };
        let ep = bincode::serialize(&enable).unwrap();
        assert_eq!(
            &ep[0..4],
            &[12, 0, 0, 0],
            "EnableFulltext discriminant changed — a variant was inserted before position 12"
        );

        // ── Variant 13: DisableFulltext { label: "A", field: "b" } ───────────
        let disable = WalRecord::DisableFulltext {
            label: "A".into(),
            field: "b".into(),
        };
        let dp = bincode::serialize(&disable).unwrap();
        assert_eq!(
            &dp[0..4],
            &[13, 0, 0, 0],
            "DisableFulltext discriminant changed — a variant was inserted before position 13"
        );

        // Roundtrip both through encode_record / decode_all.
        let mut buf = encode_record(&enable);
        buf.extend(encode_record(&disable));
        let (recs, consumed) = decode_all(&buf);
        assert_eq!(consumed, buf.len());
        assert_eq!(recs.len(), 2);
        assert_eq!(
            recs[0],
            WalRecord::EnableFulltext {
                label: "A".into(),
                field: "b".into()
            }
        );
        assert_eq!(
            recs[1],
            WalRecord::DisableFulltext {
                label: "A".into(),
                field: "b".into()
            }
        );
    }

    #[test]
    fn roundtrip_dense_id_variants_append_after_fulltext() {
        let recs = vec![
            WalRecord::Intern {
                id: 0,
                text: "Person".into(),
            },
            WalRecord::InsertNodeId {
                label: 0,
                key: "a".into(),
                props: vec![(1, Value::Int(1))],
            },
            WalRecord::SetPropId {
                id: 0,
                field: 1,
                value: Value::Int(2),
            },
            WalRecord::InsertEdgeId {
                etype: 2,
                src: 0,
                dst: 1,
            },
        ];
        for r in &recs {
            let bytes = encode_record(r);
            let (got, n) = decode_all(&bytes);
            assert_eq!(n, bytes.len());
            assert_eq!(got, vec![r.clone()]);
        }
        let p = bincode::serialize(&recs[1]).unwrap();
        assert_eq!(&p[0..4], &[14, 0, 0, 0], "InsertNodeId discriminant is 14");
        let p = bincode::serialize(&recs[2]).unwrap();
        assert_eq!(&p[0..4], &[15, 0, 0, 0], "SetPropId discriminant is 15");
        let p = bincode::serialize(&recs[3]).unwrap();
        assert_eq!(&p[0..4], &[16, 0, 0, 0], "InsertEdgeId discriminant is 16");
        let p = bincode::serialize(&recs[0]).unwrap();
        assert_eq!(&p[0..4], &[17, 0, 0, 0], "Intern discriminant is 17");
    }

    #[test]
    fn nested_batch_decode_is_treated_as_corrupt() {
        // Manually encode a Batch whose payload contains a nested Batch by
        // serializing the raw bincode bytes, bypassing encode_record's assert.
        let inner_batch = WalRecord::Batch(vec![WalRecord::DeleteNode { key: "z".into() }]);
        let outer = WalRecord::Batch(vec![inner_batch]);

        // Encode the payload without the debug assert by calling bincode directly.
        let payload = bincode::serialize(&outer).unwrap();
        let crc = crc32fast::hash(&payload);
        let mut frame = Vec::with_capacity(8 + payload.len());
        frame.extend((payload.len() as u32).to_le_bytes());
        frame.extend(crc.to_le_bytes());
        frame.extend(&payload);

        // Prepend a valid record so we can verify the stop position.
        let good = encode_record(&WalRecord::DeleteNode { key: "good".into() });
        let good_len = good.len();
        let mut bytes = good;
        bytes.extend(&frame);

        let (recs, consumed) = decode_all(&bytes);
        assert_eq!(recs.len(), 1);
        assert_eq!(recs[0], WalRecord::DeleteNode { key: "good".into() });
        assert_eq!(
            consumed, good_len,
            "stops cleanly before the nested-batch frame"
        );
    }
}