//! CUBELinux-2 record store over CZYX coordinates (PDF Package 1).
//!
//! Built NEW from the PDF spec; not recycled from the prior `/home/CUBELinux`
//! build. The prior build used a 3-axis `u64` point + `SpaceId`; this store
//! is keyed by the PDF's [`Czyx`] coordinate directly.
//!
//! Scope of this file: Package 1 only — a coordinate type, a tri-channel
//! codec, a header, and a minimal store abstraction that can later be backed
//! by a log-structured / RocksDB store. The PDF's later packages (cubefs,
//! cubevm, cubecrypt, cubeai) are out of scope for this hardware and are not
//! implemented here.

#![forbid(unsafe_code)]
#![warn(missing_docs)]

use cubecoords::{CubeHeader, Czyx};
use std::collections::HashMap;

/// A backend that maps CZYX coordinates to byte payloads.
///
/// Decision: trait takes `Czyx` by value for `put`/`delete` (small, `Copy`)
/// and by reference for `get`, matching the PDF signature while staying
/// allocation-light. Revisit if a future backend needs the whole key moved.
pub trait CubeBackend {
    /// Store `value` at `key`.
    fn put(&mut self, key: Czyx, value: Vec<u8>);
    /// Store `value` at `key`, returning `Err` if the write cannot be
    /// durably committed (e.g. the daemon socket is unreachable for
    /// `DaemonBackend`). The default implementation ignores the result so
    /// existing backends stay source-compatible; `DaemonBackend` overrides it.
    /// The FUSE write path calls this variant and surfaces `Err` as `EIO`.
    fn put_checked(&mut self, key: Czyx, value: Vec<u8>) -> Result<(), String> {
        self.put(key, value);
        Ok(())
    }
    /// Fetch the value at `key`, if present.
    fn get(&self, key: &Czyx) -> Option<Vec<u8>>;
    /// Remove the value at `key`.
    fn delete(&mut self, key: &Czyx);

    /// Optional scanning primitive (PDF Package 2: "plus optional scanning
    /// primitives"). Returns every coordinate currently present.
    ///
    /// Decision: this is a provided method returning an empty `Vec` by
    /// default so existing backends stay source-compatible, and so a backend
    /// that cannot enumerate cheaply (a remote/blind KV) can honestly report
    /// "no enumeration" instead of lying. `cubefs` needs enumeration to build
    /// directory listings, and documents that requirement at its own API.
    fn keys(&self) -> Vec<Czyx> {
        Vec::new()
    }

    /// Coordinates whose `C` (and optionally `Z`, `Y`) prefix matches.
    ///
    /// Provided in terms of [`CubeBackend::keys`]; a real on-disk backend
    /// should override this with a range scan over the packed `u32` key,
    /// which is prefix-ordered because `pack_u32` puts `C` in the high byte.
    fn scan_prefix(&self, c: u8, z: Option<u8>, y: Option<u8>) -> Vec<Czyx> {
        self.keys()
            .into_iter()
            .filter(|k| {
                k.c == c
                    && z.map(|zz| k.z == zz).unwrap_or(true)
                    && y.map(|yy| k.y == yy).unwrap_or(true)
            })
            .collect()
    }
}

/// In-memory backend backed by a `HashMap<u32, Vec<u8>>` keyed by the packed
/// `u32` form of [`Czyx`].
///
/// Decision: packs to `u32` (not a 4-tuple key) so the map layout matches the
/// PDF's `HashMap<u32, Vec<u8>>` example exactly and stays cheap. A production
/// backend would replace this with the on-disk store.
#[derive(Clone)]
pub struct HashBackend(pub HashMap<u32, Vec<u8>>);

impl HashBackend {
    /// Empty backend.
    pub fn new() -> Self {
        HashBackend(HashMap::new())
    }
}

impl Default for HashBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl CubeBackend for HashBackend {
    fn put(&mut self, key: Czyx, value: Vec<u8>) {
        self.0.insert(key.pack_u32(), value);
    }
    fn get(&self, key: &Czyx) -> Option<Vec<u8>> {
        self.0.get(&key.pack_u32()).cloned()
    }
    fn delete(&mut self, key: &Czyx) {
        self.0.remove(&key.pack_u32());
    }
    fn keys(&self) -> Vec<Czyx> {
        let mut v: Vec<Czyx> = self.0.keys().map(|k| Czyx::unpack_u32(*k)).collect();
        // Deterministic order: HashMap iteration is unordered, but callers
        // (cubefs readdir) need a stable listing.
        v.sort();
        v
    }
}

/// Type-erased backend: lets `CubeFs`/`CubeStore` hold either a `HashBackend`
/// or a `DaemonBackend` behind one concrete type, so callers (e.g.
/// `cubefs-mount`, which chooses the backend at runtime from `--socket`) don't
/// have to be generic over `B`. Forwards every call to the inner backend.
impl CubeBackend for Box<dyn CubeBackend + Send + Sync> {
    fn put(&mut self, key: Czyx, value: Vec<u8>) {
        (**self).put(key, value)
    }
    fn get(&self, key: &Czyx) -> Option<Vec<u8>> {
        (**self).get(key)
    }
    fn delete(&mut self, key: &Czyx) {
        (**self).delete(key)
    }
    fn keys(&self) -> Vec<Czyx> {
        (**self).keys()
    }
    fn scan_prefix(&self, c: u8, z: Option<u8>, y: Option<u8>) -> Vec<Czyx> {
        (**self).scan_prefix(c, z, y)
    }
}

/// A record store: a header + body addressed by a [`Czyx`] label.
///
/// Decision: we serialize the header and body as a single byte buffer with a
/// length-prefixed header section, rather than relying on an external
/// `bincode` dependency (keeps CUBELinux-2 dependency-free at Package 1).
/// The header is length-prefixed so the body boundary is recoverable without
/// a fixed schema — this is the "evolve toward explicit C/Z/Y/X-mapped flag
/// bytes" step the PDF mentions, done inline.
#[derive(Clone)]
pub struct CubeStore<B: CubeBackend> {
    backend: B,
}

/// On-wire layout of a stored record:
///   `[u32 header_len][header bytes][body bytes]`
/// Header bytes = JSON of `CubeHeader`. Decision: JSON (via a tiny manual
/// serializer-free path) is overkill; we instead use a simple, stable
/// binary form below. (Documented: JSON was considered; a compact binary
/// encoding is used to avoid a serde dependency at Package 1.)
mod record_codec {
    use cubecoords::{CubeHeader, Czyx};

    // Compact, dependency-free encoding of the header.
    // Fields are written in a fixed tag-length-value stream so unknown
    // future fields can be skipped on read. Tag byte + (optional) length +
    // payload.
    //
    // Tags:
    //   1 title      (utf8)
    //   2 doc_type   (utf8)
    //   3 created_at (u64 le)
    //   4 size_bytes (u64 le)
    //   5 owner_local_user (utf8)
    //   6 owner_remote_user (utf8)
    //   7 linked_records: u16 count, then count*(4*u8) CZYX bytes
    //   8 total_accesses (u64 le)
    //   9 total_remote_accesses (u64 le)
    //  10 last_access (u64 le)
    //  11 last_remote_access (u64 le)
    //  14 word_flags (u16 le) — per-word 16-bit tri-channel WordFlags
    //  13 path (utf8) — POSIX path metatag (the original filesystem path a
    //     record was migrated from). Carries the directory hierarchy as DATA
    //     so the associative query layer can reconstruct nesting from flags.
    //  (Tag 12, raw flag bits, is documented at its emit site below.)

    pub fn encode_header(h: &CubeHeader) -> Vec<u8> {
        let mut out = Vec::new();
        if let Some(t) = &h.title {
            put_utf8(&mut out, 1, t);
        }
        if let Some(d) = &h.doc_type {
            put_utf8(&mut out, 2, d);
        }
        if let Some(c) = h.created_at {
            put_u64(&mut out, 3, c);
        }
        if let Some(s) = h.size_bytes {
            put_u64(&mut out, 4, s);
        }
        if let Some(o) = &h.owner_local_user {
            put_utf8(&mut out, 5, o);
        }
        if let Some(o) = &h.owner_remote_user {
            put_utf8(&mut out, 6, o);
        }
        if !h.linked_records.is_empty() {
            out.push(7);
            out.extend_from_slice(&(h.linked_records.len() as u16).to_le_bytes());
            for r in &h.linked_records {
                out.push(r.c);
                out.push(r.z);
                out.push(r.y);
                out.push(r.x);
            }
        }
        if h.total_accesses != 0 {
            put_u64(&mut out, 8, h.total_accesses);
        }
        if h.total_remote_accesses != 0 {
            put_u64(&mut out, 9, h.total_remote_accesses);
        }
        if let Some(a) = h.last_access {
            put_u64(&mut out, 10, a);
        }
        if let Some(a) = h.last_remote_access {
            put_u64(&mut out, 11, a);
        }
        if let Some(p) = &h.path {
            put_utf8(&mut out, 13, p);
        }
        // Tag 12: raw flag bits. Serializes out-of-band/spare bits (e.g.
        // `cubecrypt::HEADER_FLAG_ENCRYPTED`) that are not derived from
        // structured fields, so they survive an encode/decode round-trip.
        if h.flags.bits() != 0 {
            out.push(12);
            out.extend_from_slice(&h.flags.bits().to_le_bytes());
        }
        // Tag 14: per-word 16-bit tri-channel WordFlags field (see cubecoords).
        if h.word_flags.bits() != 0 {
            out.push(14);
            out.extend_from_slice(&h.word_flags.bits().to_le_bytes());
        }
        out
    }

    pub fn decode_header(mut b: &[u8]) -> Option<CubeHeader> {
        let mut h = CubeHeader::new();
        while !b.is_empty() {
            let tag = b[0];
            b = &b[1..];
            match tag {
                1 => {
                    let (v, rest) = take_utf8(b)?;
                    h.title = Some(v);
                    b = rest;
                }
                2 => {
                    let (v, rest) = take_utf8(b)?;
                    h.doc_type = Some(v);
                    b = rest;
                }
                3 => {
                    let (v, rest) = take_u64(b)?;
                    h.created_at = Some(v);
                    b = rest;
                }
                4 => {
                    let (v, rest) = take_u64(b)?;
                    h.size_bytes = Some(v);
                    b = rest;
                }
                5 => {
                    let (v, rest) = take_utf8(b)?;
                    h.owner_local_user = Some(v);
                    b = rest;
                }
                6 => {
                    let (v, rest) = take_utf8(b)?;
                    h.owner_remote_user = Some(v);
                    b = rest;
                }
                7 => {
                    if b.len() < 2 {
                        return None;
                    }
                    let n = u16::from_le_bytes([b[0], b[1]]) as usize;
                    b = &b[2..];
                    if b.len() < n * 4 {
                        return None;
                    }
                    for _ in 0..n {
                        let c = b[0];
                        let z = b[1];
                        let y = b[2];
                        let x = b[3];
                        h.linked_records.push(Czyx::new(c, z, y, x));
                        b = &b[4..];
                    }
                }
                8 => {
                    let (v, rest) = take_u64(b)?;
                    h.total_accesses = v;
                    b = rest;
                }
                9 => {
                    let (v, rest) = take_u64(b)?;
                    h.total_remote_accesses = v;
                    b = rest;
                }
                10 => {
                    let (v, rest) = take_u64(b)?;
                    h.last_access = Some(v);
                    b = rest;
                }
                11 => {
                    let (v, rest) = take_u64(b)?;
                    h.last_remote_access = Some(v);
                    b = rest;
                }
                13 => {
                    let (v, rest) = take_utf8(b)?;
                    h.path = Some(v);
                    b = rest;
                }
                12 => {
                    if b.len() < 2 {
                        return None;
                    }
                    let raw = u16::from_le_bytes([b[0], b[1]]);
                    h.flags = cubecoords::HeaderFlags::flags_from_bits(raw);
                    b = &b[2..];
                }
                14 => {
                    if b.len() < 2 {
                        return None;
                    }
                    let raw = u16::from_le_bytes([b[0], b[1]]);
                    h.word_flags = cubecoords::WordFlags::from_bits(raw);
                    b = &b[2..];
                }
                _ => return None, // unknown tag -> reject (strict at Package 1)
            }
        }
        h.refresh_flags();
        Some(h)
    }

    fn put_u64(out: &mut Vec<u8>, tag: u8, v: u64) {
        out.push(tag);
        out.extend_from_slice(&v.to_le_bytes());
    }
    fn put_utf8(out: &mut Vec<u8>, tag: u8, s: &str) {
        let bytes = s.as_bytes();
        out.push(tag);
        out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
        out.extend_from_slice(bytes);
    }
    fn take_u64(b: &[u8]) -> Option<(u64, &[u8])> {
        if b.len() < 8 {
            return None;
        }
        let mut a = [0u8; 8];
        a.copy_from_slice(&b[..8]);
        Some((u64::from_le_bytes(a), &b[8..]))
    }
    fn take_utf8(b: &[u8]) -> Option<(String, &[u8])> {
        if b.len() < 4 {
            return None;
        }
        let mut len = [0u8; 4];
        len.copy_from_slice(&b[..4]);
        let n = u32::from_le_bytes(len) as usize;
        let rest = &b[4..];
        if rest.len() < n {
            return None;
        }
        let s = String::from_utf8(rest[..n].to_vec()).ok()?;
        Some((s, &rest[n..]))
    }
}

impl<B: CubeBackend> CubeStore<B> {
    /// Wrap a backend.
    pub fn new(backend: B) -> Self {
        CubeStore { backend }
    }

    /// Store `header` + `body` at `label`.
    pub fn put_record(&mut self, label: Czyx, header: &CubeHeader, body: &[u8]) {
        let hdr_bytes = record_codec::encode_header(header);
        let mut buf = Vec::with_capacity(4 + hdr_bytes.len() + body.len());
        buf.extend_from_slice(&(hdr_bytes.len() as u32).to_le_bytes());
        buf.extend_from_slice(&hdr_bytes);
        buf.extend_from_slice(body);
        self.backend.put(label, buf)
    }

    /// Like [`CubeStore::put_record`] but returns `Err` if the backend cannot
    /// durably commit (e.g. `DaemonBackend` with an unreachable socket). The
    /// FUSE write path uses this variant so a failed daemon write becomes
    /// `EIO` instead of a silently-dropped write.
    pub fn put_record_checked(
        &mut self,
        label: Czyx,
        header: &CubeHeader,
        body: &[u8],
    ) -> Result<(), String> {
        let hdr_bytes = record_codec::encode_header(header);
        let mut buf = Vec::with_capacity(4 + hdr_bytes.len() + body.len());
        buf.extend_from_slice(&(hdr_bytes.len() as u32).to_le_bytes());
        buf.extend_from_slice(&hdr_bytes);
        buf.extend_from_slice(body);
        self.backend.put_checked(label, buf)
    }

    /// Fetch and split a record into `(header, body)`.
    ///
    /// Records written through the record path carry a TLV envelope
    /// (`u32 header-len | header | body`). Records written through the *raw*
    /// path (`put_raw` / the daemon's `rawput` verb, used by OS-layer services)
    /// carry no envelope at all.
    ///
    /// Historically a non-enveloped payload made this return `None`, which
    /// callers such as `cubefs`'s `getattr`/`read` and the daemon's `stat` verb
    /// translate into "size 0 / empty file". A record holding real bytes was
    /// therefore *silently invisible* through the filesystem while `rawget`
    /// happily returned its contents — the OS-in-CUBE migration hit exactly
    /// this (2026-08-13): every `rawput` OS record listed as a 0-byte file.
    ///
    /// Losing data silently is never the right failure mode, so a payload that
    /// is not a well-formed envelope is now surfaced as a raw body under a
    /// synthesized header. `rawput` data becomes readable through the
    /// filesystem, and no caller has to special-case the two write paths.
    pub fn get_record(&self, label: &Czyx) -> Option<(CubeHeader, Vec<u8>)> {
        let raw = self.backend.get(label)?;
        // Fall back to treating the payload as a raw (un-enveloped) body when
        // it cannot be parsed as `len | header | body`.
        let raw_fallback = |bytes: &[u8]| {
            let mut h = CubeHeader::new();
            h.size_bytes = Some(bytes.len() as u64);
            // Keep the synthesized header's flag bits consistent with its
            // fields, so associative queries (`scan_by_flag(SIZE_BYTES)`) see
            // raw records too. Only the derived low bits are recomputed; any
            // out-of-band bits already on the payload-derived header (none
            // here, since raw payloads carry no header) are preserved by
            // `refresh_flags`.
            h.refresh_flags();
            Some((h, bytes.to_vec()))
        };
        if raw.len() < 4 {
            return raw_fallback(&raw);
        }
        let mut len = [0u8; 4];
        len.copy_from_slice(&raw[..4]);
        let hlen = u32::from_le_bytes(len) as usize;
        if raw.len() < 4 + hlen {
            return raw_fallback(&raw);
        }
        match record_codec::decode_header(&raw[4..4 + hlen]) {
            Some(hdr) => {
                let body = raw[4 + hlen..].to_vec();
                Some((hdr, body))
            }
            None => raw_fallback(&raw),
        }
    }

    /// Raw backend write (non-record payloads, e.g. ACL/xattr/volume buckets).
    pub fn put_raw(&mut self, key: Czyx, value: Vec<u8>) {
        self.backend.put(key, value)
    }

    /// Like [`CubeStore::put_raw`] but returns `Err` if the backend cannot
    /// durably commit. Used by the FUSE write path for best-effort metadata;
    /// callers may `.ok()` it or surface the error.
    pub fn put_raw_checked(&mut self, key: Czyx, value: Vec<u8>) -> Result<(), String> {
        self.backend.put_checked(key, value)
    }

    /// Raw backend get.
    pub fn get_raw(&self, key: &Czyx) -> Option<Vec<u8>> {
        self.backend.get(key)
    }
    /// Raw backend delete.
    pub fn delete_raw(&mut self, key: &Czyx) {
        self.backend.delete(key);
    }

    /// Every coordinate present in the backend (requires a backend that
    /// implements [`CubeBackend::keys`]).
    pub fn keys(&self) -> Vec<Czyx> {
        self.backend.keys()
    }

    /// Coordinates under a `C`/`Z`/`Y` prefix.
    pub fn scan_prefix(&self, c: u8, z: Option<u8>, y: Option<u8>) -> Vec<Czyx> {
        self.backend.scan_prefix(c, z, y)
    }

    /// PDF Package 2 API: link `src` to `dst` by appending `dst` to `src`'s
    /// `linked_records` and refreshing the association flag.
    ///
    /// Decision: the association is stored one-way in the source header (as
    /// the PDF's "association flags" describe), and reverse lookup is done by
    /// scanning (see [`CubeStore::linked_to`]). Storing a reverse index would
    /// double-write every association and risk divergence; scanning is cheap
    /// against the packed-u32 key space and always consistent.
    /// Returns `false` if `src` does not exist.
    pub fn associate(&mut self, src: Czyx, dst: Czyx) -> bool {
        let Some((mut h, body)) = self.get_record(&src) else {
            return false;
        };
        if !h.linked_records.contains(&dst) {
            h.linked_records.push(dst);
        }
        h.refresh_flags();
        self.put_record(src, &h, &body);
        true
    }

    /// PDF Package 2 API: "all records linked to X" — every coordinate whose
    /// header lists `target` in its `linked_records`.
    pub fn linked_to(&self, target: &Czyx) -> Vec<Czyx> {
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| h.linked_records.contains(target))
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// PDF Package 2 API: `scan_by_flag` — associative storage lookup.
    ///
    /// Returns every coordinate whose decoded [`CubeHeader`] carries `flag`
    /// set. This is the "query by metatag/flag, not by path" primitive the
    /// source PDF describes (§"associative storage"): records are addressed
    /// by *what they are* (a flag bit) rather than *where they live* (a
    /// coordinate path). `cube-os-*` services that emit typed records
    /// (e.g. a log line tagged `doc_type = "klog"`) become discoverable by
    /// event type without knowing their CZYX address in advance.
    ///
    /// Records written through the RAW path (`rawput`, used by the OS layers)
    /// carry only a synthesized `size_bytes` header, so they will NOT match
    /// a content flag unless the writer also set one. The associative query
    /// is therefore most powerful when records are written through
    /// [`CubeStore::put_record`] with a populated [`CubeHeader`].
    pub fn scan_by_flag(&self, flag: u16) -> Vec<Czyx> {
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| h.flags.has(flag))
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// Query records by per-word flag (the 16-bit WordFlags metadata stripe
    /// carried in the header, tag 14). Parallel to [`scan_by_flag`] but on
    /// `word_flags`, so per-word metadata (encrypted/compressed/stego/type/
    /// arrangement/continuation) is queryable without decoding the body.
    pub fn scan_by_word_flag(&self, flag: u16) -> Vec<Czyx> {
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| h.word_flags.has(flag))
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// PDF Package 2 API: `scan_by_type` — query records by `doc_type`
    /// (the "event type" / file-extension analogue the PDF calls Flag 2).
    ///
    /// This is the concrete "log lookup by query" the OS layers want: instead
    /// of addressing `/cubefs/c200/z004/...` directly, you ask "every record
    /// whose doc_type is `klog`" and get back all matching coordinates. The
    /// comparison is exact-match (case-sensitive) against the header field.
    pub fn scan_by_type(&self, doc_type: &str) -> Vec<Czyx> {
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| h.doc_type.as_deref() == Some(doc_type))
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// Return `(label, header, body)` for every record matching `flag`.
    /// Convenience wrapper over [`CubeStore::scan_by_flag`] that also pulls
    /// the decoded payload so a caller (e.g. a `cubelog` query tool) can
    /// present the matching records directly.
    pub fn query_by_flag(&self, flag: u16) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
        self.scan_by_flag(flag)
            .into_iter()
            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
            .collect()
    }

    /// Return `(label, header, body)` for every record whose `doc_type`
    /// matches. See [`CubeStore::scan_by_type`] for the matching semantics.
    pub fn query_by_type(&self, doc_type: &str) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
        self.scan_by_type(doc_type)
            .into_iter()
            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
            .collect()
    }

    /// PDF Package 2 / OS-in-CUBE: reconstruct a *directory hierarchy from
    /// metatags*. Returns every coordinate whose header carries a `path`
    /// metatag equal to `path` (exact) — i.e. "the file at this path".
    ///
    /// Combined with [`CubeStore::scan_by_path_prefix`], this lets the cube
    /// answer "everything under /etc" WITHOUT the filesystem supporting
    /// recursive nesting: the path is stored as a flag-addressed field, not
    /// as an inode tree. This is the source PDF's prescribed model ("operate
    /// on CZYX records and Null-space flags rather than paths and inodes").
    pub fn scan_by_path(&self, path: &str) -> Vec<Czyx> {
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| h.path.as_deref() == Some(path))
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// "List a directory": every record whose `path` metatag is *under* the
    /// given directory prefix (e.g. `scan_by_path_prefix("/etc")` returns
    /// `/etc/passwd`, `/etc/network/interfaces`, ...). Unlike a real FS, the
    /// nesting is a shared prefix on the `path` metatag of several records
    /// (the source PDF: "operate on CZYX records and Null-space flags rather
    /// than paths and inodes") — there is no half-coordinate directory entry.
    ///
    /// A query matches BOTH an exact leaf (`/etc/hostname`) and any descendant
    /// (`/etc/hostname` and `/etc/passwd` both answer `/etc`), because a path
    /// is just a string: `/etc/hostname` == dir, and `/etc/passwd` starts with
    /// `dir + '/'`.
    pub fn scan_by_path_prefix(&self, dir: &str) -> Vec<Czyx> {
        let sep = format!("{dir}/");
        let mut out: Vec<Czyx> = self
            .backend
            .keys()
            .into_iter()
            .filter(|k| {
                self.get_record(k)
                    .map(|(h, _)| match &h.path {
                        Some(p) => p == dir || p.starts_with(&sep),
                        None => false,
                    })
                    .unwrap_or(false)
            })
            .collect();
        out.sort();
        out
    }

    /// `(label, header, body)` for [`CubeStore::scan_by_path`].
    pub fn query_by_path(&self, path: &str) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
        self.scan_by_path(path)
            .into_iter()
            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
            .collect()
    }

    /// `(label, header, body)` for [`CubeStore::scan_by_path_prefix`].
    pub fn query_by_path_prefix(&self, dir: &str) -> Vec<(Czyx, CubeHeader, Vec<u8>)> {
        self.scan_by_path_prefix(dir)
            .into_iter()
            .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b)))
            .collect()
    }
}

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

    #[test]
    fn record_roundtrip() {
        let mut store = CubeStore::new(HashBackend::new());
        let mut h = CubeHeader::new();
        h.title = Some("memory".into());
        h.doc_type = Some("note".into());
        h.created_at = Some(1700000000);
        h.size_bytes = Some(5);
        h.linked_records.push(Czyx::new(1, 2, 3, 4));
        h.refresh_flags();

        let label = Czyx::new(1, 10, 20, 30);
        store.put_record(label, &h, b"hello");
        let (rh, body) = store.get_record(&label).unwrap();
        assert_eq!(body, b"hello");
        assert_eq!(rh.title.as_deref(), Some("memory"));
        assert_eq!(rh.doc_type.as_deref(), Some("note"));
        assert_eq!(rh.created_at, Some(1700000000));
        assert_eq!(rh.size_bytes, Some(5));
        assert_eq!(rh.linked_records, vec![Czyx::new(1, 2, 3, 4)]);
        assert!(rh.flags.has(cubecoords::HeaderFlags::HAS_ASSOCIATIONS));
    }


    #[test]
    fn word_flags_roundtrip() {
        let mut h = CubeHeader::new();
        h.word_flags = cubecoords::WordFlags::from_bits(
            cubecoords::WordFlags::ENCRYPTED
                | cubecoords::WordFlags::CONTINUATION
                | (cubecoords::WordFlags::TYPE_MASK & 0b11),
        );
        let enc = super::record_codec::encode_header(&h);
        let dec = super::record_codec::decode_header(&enc).unwrap();
        assert_eq!(dec.word_flags.bits(), h.word_flags.bits(), "word_flags must round-trip");
        // refresh_flags recomputes only the per-RECORD HeaderFlags; the separate
        // per-word word_flags field must be preserved untouched.
        let mut dec2 = dec;
        dec2.refresh_flags();
        assert_eq!(dec2.word_flags.bits(), h.word_flags.bits());
    }

    #[test]
    fn missing_record_is_none() {
        let store = CubeStore::new(HashBackend::new());
        assert!(store.get_record(&Czyx::new(9, 9, 9, 9)).is_none());
    }

    #[test]
    fn null_coord_is_distinct_key() {
        let mut store = CubeStore::new(HashBackend::new());
        store.put_raw(Czyx::new(0, 0, 0, 0), vec![1]);
        store.put_raw(Czyx::new(0, 0, 0, 1), vec![2]);
        assert_eq!(store.get_raw(&Czyx::new(0, 0, 0, 0)), Some(vec![1]));
        assert_eq!(store.get_raw(&Czyx::new(0, 0, 0, 1)), Some(vec![2]));
    }

    /// Regression (2026-08-13, OS-in-CUBE migration): a payload written via the
    /// RAW path (`put_raw`, i.e. the daemon's `rawput` verb used by OS-layer
    /// services) must still be READABLE through `get_record`, because that is
    /// what `cubefs` getattr/read and the `stat` verb go through. Before the
    /// fix these records reported size 0 and read back empty — real bytes were
    /// silently invisible through the filesystem.
    #[test]
    fn get_record_surfaces_raw_unenveloped_payloads() {
        let mut store = CubeStore::new(HashBackend::new());
        let coord = Czyx::new(200, 70, 1, 1);
        let payload = b"CUBELINUX OS PROCESS SNAPSHOT\nprocs: 118\n".to_vec();
        store.put_raw(coord, payload.clone());

        let (hdr, body) = store
            .get_record(&coord)
            .expect("raw payload must be visible as a record, not vanish");
        assert_eq!(body, payload, "body must round-trip byte-for-byte");
        assert_eq!(
            hdr.size_bytes,
            Some(payload.len() as u64),
            "synthesized header must report the true size so getattr is correct"
        );

        // A short payload (< 4 bytes, cannot even hold a length prefix) is the
        // other edge the old code dropped: counters like "118" land here.
        let short = Czyx::new(200, 70, 2, 1);
        store.put_raw(short, b"118".to_vec());
        let (h2, b2) = store.get_record(&short).expect("short raw payload visible");
        assert_eq!(b2, b"118");
        assert_eq!(h2.size_bytes, Some(3));
    }

    /// The envelope path must be unaffected by the raw fallback: a properly
    /// stored record still decodes its real header (not a synthesized one).
    #[test]
    fn get_record_still_prefers_the_real_envelope() {
        let mut store = CubeStore::new(HashBackend::new());
        let coord = Czyx::new(201, 5, 1, 1);
        let mut hdr = CubeHeader::new();
        hdr.title = Some("real-record".to_string());
        store.put_record(coord, &hdr, b"payload");

        let (got, body) = store.get_record(&coord).expect("enveloped record");
        assert_eq!(body, b"payload");
        assert_eq!(
            got.title.as_deref(),
            Some("real-record"),
            "must decode the true header, not fall back to raw"
        );
    }

    /// PDF Package 2: `scan_by_flag` finds records by flag bit, independent of
    /// their coordinate address. A record tagged by `DOC_TYPE` must surface
    /// when queried for that flag and be absent otherwise.
    #[test]
    fn scan_by_flag_finds_typed_records() {
        let mut store = CubeStore::new(HashBackend::new());

        // A "klog" typed event record.
        let mut klog = CubeHeader::new();
        klog.doc_type = Some("klog".into());
        klog.refresh_flags();
        store.put_record(Czyx::new(200, 4, 2, 1), &klog, b"kernel: eth0 up");

        // An untyped raw record (the common OS-layer case).
        store.put_raw(Czyx::new(200, 70, 1, 1), b"118".to_vec());

        let by_doc_type = store.scan_by_flag(cubecoords::HeaderFlags::DOC_TYPE);
        assert_eq!(by_doc_type, vec![Czyx::new(200, 4, 2, 1)]);

        // The raw record carries only a synthesized size flag, so it must NOT
        // match DOC_TYPE.
        assert!(!by_doc_type.contains(&Czyx::new(200, 70, 1, 1)));

        let by_size = store.scan_by_flag(cubecoords::HeaderFlags::SIZE_BYTES);
        assert!(
            !by_size.is_empty(),
            "synthesized raw headers carry SIZE_BYTES"
        );
    }

    #[test]
    fn scan_by_word_flag_finds_records() {
        let mut store = CubeStore::new(HashBackend::new());
        let mut h = CubeHeader::new();
        h.word_flags = cubecoords::WordFlags::from_bits(cubecoords::WordFlags::ENCRYPTED);
        store.put_record(Czyx::new(200, 5, 1, 1), &h, b"secret body");

        let mut h2 = CubeHeader::new();
        h2.doc_type = Some("plain".into());
        h2.refresh_flags();
        store.put_record(Czyx::new(200, 5, 2, 2), &h2, b"plain body");

        let enc = store.scan_by_word_flag(cubecoords::WordFlags::ENCRYPTED);
        assert_eq!(enc, vec![Czyx::new(200, 5, 1, 1)]);
        assert!(!enc.contains(&Czyx::new(200, 5, 2, 2)));
    }

    /// `scan_by_type` is the concrete "log lookup by query" — pull every record
    /// of a given event type without knowing its coordinate.
    #[test]
    fn scan_by_type_event_lookup() {
        let mut store = CubeStore::new(HashBackend::new());
        let mut a = CubeHeader::new();
        a.doc_type = Some("klog".into());
        let mut b = CubeHeader::new();
        b.doc_type = Some("klog".into());
        let mut c = CubeHeader::new();
        c.doc_type = Some("state".into());
        store.put_record(Czyx::new(200, 4, 2, 1), &a, b"k1");
        store.put_record(Czyx::new(200, 4, 2, 2), &b, b"k2");
        store.put_record(Czyx::new(200, 1, 1, 1), &c, b"s1");

        let klogs = store.scan_by_type("klog");
        assert_eq!(klogs.len(), 2);
        assert!(klogs.contains(&Czyx::new(200, 4, 2, 1)));
        assert!(klogs.contains(&Czyx::new(200, 4, 2, 2)));

        // query_by_type returns the decoded payloads too.
        let klogs_full = store.query_by_type("klog");
        assert_eq!(klogs_full.len(), 2);
        assert!(klogs_full.iter().any(|(_, _, b)| b == b"k1"));
        assert!(klogs_full.iter().any(|(_, _, b)| b == b"k2"));

        assert_eq!(store.scan_by_type("state"), vec![Czyx::new(200, 1, 1, 1)]);
        assert!(store.scan_by_type("nonexistent").is_empty());
    }

    /// Path metatags reconstruct a nested directory hierarchy WITHOUT the FS
    /// needing recursive inode trees (source PDF: "operate on CZYX records and
    /// Null-space flags rather than paths and inodes"). `/etc/passwd` and
    /// `/etc/network/interfaces` live at unrelated coordinates but share the
    /// `/etc` prefix, so `scan_by_path_prefix("/etc")` finds both.
    #[test]
    fn scan_by_path_prefix_lists_directory() {
        let mut store = CubeStore::new(HashBackend::new());
        let mut mk = |c: u8, path: &str, dt: &str| {
            let mut h = CubeHeader::new();
            h.path = Some(path.to_string());
            h.doc_type = Some(dt.to_string());
            h.refresh_flags();
            store.put_record(Czyx::new(200, 1, c, 1), &h, b"body");
        };
        mk(1, "/etc/passwd", "file");
        mk(2, "/etc/network/interfaces", "file");
        mk(3, "/etc/hosts", "file");
        mk(4, "/usr/bin/ls", "file");

        // Exact match.
        assert_eq!(
            store.scan_by_path("/etc/passwd"),
            vec![Czyx::new(200, 1, 1, 1)]
        );

        // Directory listing via metatag prefix. Bare "/etc" matches every
        // record whose path is "/etc" or starts with "/etc/" (an exact leaf
        // like "/etc/hostname" answers "/etc", and "/etc/passwd" is a child).
        let etc = store.scan_by_path_prefix("/etc");
        assert_eq!(etc.len(), 3);
        assert!(etc.contains(&Czyx::new(200, 1, 1, 1)));
        assert!(etc.contains(&Czyx::new(200, 1, 2, 1)));
        assert!(etc.contains(&Czyx::new(200, 1, 3, 1)));
        assert!(
            !etc.contains(&Czyx::new(200, 1, 4, 1)),
            "/usr must not appear under /etc"
        );

        // An exact leaf answers its own parent directory query.
        assert_eq!(
            store.scan_by_path_prefix("/etc/hosts"),
            vec![Czyx::new(200, 1, 3, 1)]
        );

        // Empty prefix = all paths.
        assert_eq!(store.scan_by_path_prefix("").len(), 4);

        // The path flag bit is set so the record is also flag-discoverable.
        let by_path_flag = store.scan_by_flag(cubecoords::HeaderFlags::HAS_PATH);
        assert_eq!(by_path_flag.len(), 4);
    }

    // ---- HashBackend trait-level tests (PDF spec: HashMap<u32, Vec<u8>>) ----

    #[test]
    fn hash_backend_put_get_delete_roundtrip() {
        let mut b = HashBackend::new();
        let k = Czyx::new(10, 20, 30, 40);
        b.put(k, b"payload".to_vec());
        assert_eq!(b.get(&k), Some(b"payload".to_vec()));
        b.delete(&k);
        assert_eq!(b.get(&k), None);
    }

    #[test]
    fn hash_backend_overwrite_replaces_value() {
        let mut b = HashBackend::new();
        let k = Czyx::new(0, 1, 2, 3);
        b.put(k, b"old".to_vec());
        b.put(k, b"new".to_vec());
        assert_eq!(b.get(&k), Some(b"new".to_vec()));
    }

    #[test]
    fn hash_backend_keys_sorted_order() {
        let mut b = HashBackend::new();
        for c in 1u8..=3 {
            b.put(Czyx::new(50, c, 10, 1), vec![c]);
        }
        let keys = b.keys();
        assert_eq!(keys.len(), 3);
        // HashMap iteration is unordered; keys() must sort for stable output.
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(keys, sorted);
        for k in &keys {
            assert_eq!(k.c, 50);
        }
    }

    #[test]
    fn hash_backend_scan_prefix_filters_correctly() {
        let mut b = HashBackend::new();
        b.put(Czyx::new(1, 2, 3, 4), b"a".to_vec());
        b.put(Czyx::new(1, 9, 9, 9), b"b".to_vec());
        b.put(Czyx::new(2, 0, 0, 0), b"c".to_vec());
        b.put(Czyx::new(1, 2, 9, 9), b"d".to_vec());

        // C=1 only.
        let c1 = b.scan_prefix(1, None, None);
        assert_eq!(c1.len(), 3);
        for k in &c1 {
            assert_eq!(k.c, 1);
        }

        // C=1, Z=2 only.
        let c1z2 = b.scan_prefix(1, Some(2), None);
        assert_eq!(c1z2.len(), 2);
        for k in &c1z2 {
            assert_eq!(k.c, 1);
            assert_eq!(k.z, 2);
        }

        // C=2 only.
        let c2 = b.scan_prefix(2, None, None);
        assert_eq!(c2, vec![Czyx::new(2, 0, 0, 0)]);

        // Non-existent C.
        assert!(b.scan_prefix(99, None, None).is_empty());
    }

    #[test]
    fn hash_backend_empty_keys_and_scan() {
        let b = HashBackend::new();
        assert!(b.keys().is_empty());
        assert!(b.scan_prefix(0, None, None).is_empty());
    }

    #[test]
    fn hash_backend_null_coord_is_valid_key() {
        let mut b = HashBackend::new();
        let null = Czyx::new(0, 0, 0, 0);
        b.put(null, b"null-cube".to_vec());
        assert_eq!(b.get(&null), Some(b"null-cube".to_vec()));
        b.delete(&null);
        assert_eq!(b.get(&null), None);

        // Null cube (0,0,0,0) is distinct from a nearby coord.
        let near = Czyx::new(0, 0, 0, 1);
        b.put(near, b"near".to_vec());
        assert_eq!(b.get(&near), Some(b"near".to_vec()));
        assert_eq!(b.get(&null), None);
    }

    #[test]
    fn hash_backend_multiple_backends_are_independent() {
        let b1 = HashBackend::new();
        let mut b2 = HashBackend::new();
        b2.put(Czyx::new(1, 1, 1, 1), b"x".to_vec());
        assert!(b1.get(&Czyx::new(1, 1, 1, 1)).is_none());
        assert_eq!(b2.get(&Czyx::new(1, 1, 1, 1)), Some(b"x".to_vec()));
    }
}