reddb-io-server 1.36.3

use super::*;

pub(super) struct Catalog {
    pub header: f::NativePagedMetadataHeader,
    pub roots: Vec<f::NativePagedCollectionRoot>,
    references: Vec<f::NativePagedCrossRef>,
    epoch: Option<u64>,
    retired_roots: Vec<u32>,
    deferred_pages: Vec<u32>,
    pub resume: Option<f::native_resume::CatalogResume>,
    proofs: Option<Arc<f::page_proof::CatalogProofs>>,
}
impl Catalog {
    pub fn has_cross_references(&self) -> bool {
        !self.references.is_empty()
    }
    pub fn read(source: &Source) -> Result<Self> {
        let reference = source.manifest_reference;
        let root = if reference.page_id == 0 {
            1
        } else {
            reference.page_id
        };
        if root >= FIRST_ALLOCATABLE_PAGE {
            source.claim(root)?;
        }
        let page = source.raw_page(root, true)?;
        let mut digest = f::page_proof::CatalogDigest::default();
        digest.add_page(page.as_bytes());
        ensure(
            page.page_type()? == PageType::Header,
            "invalid catalog page",
        )?;
        let mut data = page.content().to_vec();
        if let Some(header) = f::decode_native_metadata_overflow_header(&data)? {
            let total = header.total_payload_bytes as usize;
            ensure(
                total > 0 && total <= 64 * 1024 * 1024,
                "catalog exceeds compaction memory budget",
            )?;
            let first = total.min(data.len() - 16);
            let mut payload = data[16..16 + first].to_vec();
            let mut next = header.next_overflow_page_id;
            while next != 0 {
                source.claim(next)?;
                let page = source.raw_page(next, true)?;
                digest.add_page(page.as_bytes());
                ensure(
                    page.page_type()? == PageType::Overflow,
                    "invalid catalog continuation",
                )?;
                let continuation =
                    f::decode_native_metadata_overflow_continuation_header(page.content())?;
                let length = continuation.chunk_bytes as usize;
                ensure(
                    length > 0
                        && length <= PAGE_SIZE - HEADER_SIZE - 8
                        && payload.len() + length <= total,
                    "invalid catalog continuation length",
                )?;
                payload.extend_from_slice(&page.content()[8..8 + length]);
                next = continuation.next_overflow_page_id;
            }
            ensure(payload.len() == total, "incomplete catalog")?;
            data = payload;
        }
        ensure(
            reference.page_id == 0 || digest.finish() == reference.digest,
            "catalog generation proof does not match the superblock",
        )?;
        let header = f::decode_native_paged_metadata_header(&data)?
            .ok_or("source is not a native paged catalog")?;
        ensure(
            matches!(header.format_version, 11 | 12),
            "unsupported native record format",
        )?;
        let mut position = f::METADATA_HEADER_BYTES;
        ensure(
            header.collection_count as usize <= data.len() / 8,
            "invalid catalog count",
        )?;
        let mut roots = Vec::new();
        let mut names = std::collections::BTreeSet::new();
        for _ in 0..header.collection_count {
            let root = f::decode_native_paged_collection_root(&data, &mut position)?;
            ensure(
                names.insert(root.collection.clone()),
                "duplicate collection",
            )?;
            roots.push(root);
        }
        let count = number(&data, position)?;
        position += 4;
        ensure(
            count as usize <= data.len().saturating_sub(position) / 21,
            "invalid reference count",
        )?;
        let mut references = Vec::new();
        for _ in 0..count {
            references.push(f::decode_native_paged_cross_ref(&data, &mut position)?);
        }
        // Offline ownership has no reader pins. Retired generations and deferred
        // frees are physical garbage, not MVCC versions in the current trees.
        let retired_roots = f::decode_native_paged_retired_trees(&data, &mut position)?;
        let deferred_pages = f::decode_native_paged_deferred_pages(&data, &mut position)?;
        let epoch = f::native_store::decode_native_paged_ownership_epoch(&data, &mut position)?;
        let proofs = f::page_proof::decode_catalog(&data, &mut position)?.map(Arc::new);
        ensure(
            proofs.is_none()
                || (source.header.required_capabilities
                    & f::file_format::FILE_CAPABILITY_TREE_PROOFS
                    != 0
                    && reference.page_id != 0),
            "tree proofs require independent catalog authority",
        )?;
        let resume = f::native_resume::decode_catalog_resume(&data)?;
        if let Some(resume) = &resume {
            let mut canonical = Vec::new();
            f::native_resume::encode_catalog_resume(&mut canonical, resume)?;
            ensure(
                data.get(position..position + canonical.len()) == Some(canonical.as_slice()),
                "unsupported/noncanonical resume trailer",
            )?;
            position += canonical.len();
        }
        ensure(
            data[position..].iter().all(|byte| *byte == 0),
            "unknown catalog extension",
        )?;
        if let Some(proofs) = &proofs {
            let live = roots
                .iter()
                .map(|root| root.root_page)
                .chain(
                    resume
                        .iter()
                        .flat_map(|resume| resume.collections.values())
                        .flat_map(|state| {
                            state.key_index_root.into_iter().chain(
                                state
                                    .secondary_indices
                                    .values()
                                    .map(|index| index.root_page),
                            )
                        }),
                )
                .filter(|root| *root != 0)
                .collect::<std::collections::BTreeSet<_>>();
            ensure(
                live.iter()
                    .chain(retired_roots.iter())
                    .all(|root| proofs.contains_key(root)),
                "published tree has no generation proof",
            )?;
            ensure(
                proofs.iter().all(|(root, proof)| {
                    (live.contains(root) || retired_roots.contains(root))
                        && *root < source.header.page_count
                        && proof.reference.page_id < source.header.page_count
                }),
                "unowned catalog tree proof",
            )?;
        }
        source.install_tree_proofs(proofs.clone())?;
        Ok(Self {
            header,
            roots,
            references,
            epoch,
            retired_roots,
            deferred_pages,
            resume,
            proofs,
        })
    }

    /// Offline readers have no pins, but retired pages still need an explicit
    /// ownership proof. Never treat arbitrary unclassified pages as garbage.
    pub fn claim_retired(&self, source: &Arc<Source>) -> Result<()> {
        for root in &self.retired_roots {
            let tree = source.tree(*root)?;
            for page in tree.retired_pages()? {
                source.claim(page)?;
            }
        }
        for page in &self.deferred_pages {
            source.claim(*page)?;
        }
        Ok(())
    }

    pub(super) fn verify_alias_has_no_references(&self, legacy: &str) -> Result<()> {
        ensure(
            self.references
                .iter()
                .all(|r| r.target_collection != legacy),
            "legacy runtime metadata has incoming references",
        )
    }

    pub fn retain_canonical_system_config(&mut self) -> Result<()> {
        let legacy_roots = self
            .roots
            .iter()
            .filter(|root| root.collection == "red_config")
            .count();
        let canonical_roots = self
            .roots
            .iter()
            .filter(|root| root.collection == "red.config")
            .count();
        ensure(
            legacy_roots == 1 && canonical_roots == 1,
            "canonical config repair requires exactly one red_config and one red.config root",
        )?;
        ensure(
            self.references
                .iter()
                .all(|reference| reference.target_collection != "red_config"),
            "legacy red_config has cross-references and cannot be retired",
        )?;
        let resume = self
            .resume
            .as_mut()
            .ok_or("canonical config repair requires a resume catalog")?;
        ensure(
            resume.collections.contains_key("red_config")
                && resume.collections.contains_key("red.config"),
            "canonical config repair requires both resume identities",
        )?;
        resume.collections.remove("red_config");
        self.roots.retain(|root| root.collection != "red_config");
        self.header.collection_count = u32::try_from(self.roots.len())?;
        Ok(())
    }

    /// Retire the obsolete physical config identity without discarding its rows.
    ///
    /// A historical store may contain only the real config tree as `red_config`,
    /// or it may also contain an empty bootstrap-created `red.config` tree. The
    /// populated legacy tree is already the complete value/history/index source,
    /// so promote its catalog identity atomically and retire an empty duplicate.
    /// A canonical-only store is already migrated. Two populated trees are
    /// ambiguous and must be reconciled by a row-aware recovery path instead of
    /// silently choosing either side.
    pub fn migrate_legacy_system_config(&mut self) -> Result<usize> {
        let legacy_roots = self
            .roots
            .iter()
            .filter(|root| root.collection == "red_config")
            .count();
        let canonical_roots = self
            .roots
            .iter()
            .filter(|root| root.collection == "red.config")
            .count();
        ensure(
            legacy_roots <= 1 && canonical_roots <= 1,
            "canonical config migration requires unique config roots",
        )?;
        ensure(
            legacy_roots + canonical_roots > 0,
            "canonical config migration requires a config root",
        )?;
        ensure(
            self.references
                .iter()
                .all(|reference| reference.target_collection != "red_config"),
            "legacy red_config has cross-references and cannot be retired",
        )?;

        let resume = self
            .resume
            .as_mut()
            .ok_or("canonical config migration requires a resume catalog")?;
        ensure(
            resume.collections.contains_key("red_config") == (legacy_roots == 1)
                && resume.collections.contains_key("red.config") == (canonical_roots == 1),
            "canonical config migration requires resume identities to match config roots",
        )?;
        if legacy_roots == 0 {
            return Ok(0);
        }
        let legacy = resume
            .collections
            .get("red_config")
            .ok_or("canonical config migration requires the red_config resume identity")?;
        let legacy_records = usize::try_from(legacy.entities)?;

        if canonical_roots == 0 {
            let legacy = resume
                .collections
                .remove("red_config")
                .ok_or("legacy config resume identity disappeared")?;
            resume.collections.insert("red.config".into(), legacy);
            let root = self
                .roots
                .iter_mut()
                .find(|root| root.collection == "red_config")
                .ok_or("legacy config root disappeared")?;
            root.collection = "red.config".into();
            return Ok(legacy_records);
        }

        let canonical = resume
            .collections
            .get("red.config")
            .ok_or("canonical config migration requires the red.config resume identity")?;

        match (legacy.entities > 0, canonical.entities > 0) {
            (true, true) => {
                return Err(
                    "both red_config and red.config contain records; refusing lossy migration"
                        .into(),
                )
            }
            (true, false) => {
                let legacy = resume
                    .collections
                    .remove("red_config")
                    .ok_or("legacy config resume identity disappeared")?;
                resume.collections.remove("red.config");
                resume.collections.insert("red.config".into(), legacy);

                self.roots.retain(|root| root.collection != "red.config");
                let root = self
                    .roots
                    .iter_mut()
                    .find(|root| root.collection == "red_config")
                    .ok_or("legacy config root disappeared")?;
                root.collection = "red.config".into();
                self.header.collection_count = u32::try_from(self.roots.len())?;
                Ok(legacy_records)
            }
            (false, _) => {
                resume.collections.remove("red_config");
                self.roots.retain(|root| root.collection != "red_config");
                self.header.collection_count = u32::try_from(self.roots.len())?;
                Ok(0)
            }
        }
    }

    pub fn encode(&self, roots: &BTreeMap<u32, u32>) -> Result<Vec<u8>> {
        self.encode_with_proofs(roots, self.proofs.as_deref())
    }

    pub fn encode_with_proofs(
        &self,
        roots: &BTreeMap<u32, u32>,
        proofs: Option<&f::page_proof::CatalogProofs>,
    ) -> Result<Vec<u8>> {
        let remap = |root: u32| -> Result<u32> {
            if root == 0 {
                Ok(0)
            } else {
                roots
                    .get(&root)
                    .copied()
                    .ok_or_else(|| "missing rewritten tree".into())
            }
        };
        let mut data = Vec::new();
        f::encode_native_paged_metadata_header(&mut data, self.header);
        for root in &self.roots {
            f::encode_native_paged_collection_root(
                &mut data,
                &root.collection,
                remap(root.root_page)?,
            );
        }
        data.extend_from_slice(&u32::try_from(self.references.len())?.to_le_bytes());
        for reference in &self.references {
            f::encode_native_paged_cross_ref(
                &mut data,
                reference.source_id,
                reference.target_id,
                reference.ref_type,
                &reference.target_collection,
            );
        }
        f::encode_native_paged_retired_trees(&mut data, &[])?;
        f::encode_native_paged_deferred_pages(&mut data, &[])?;
        if let Some(epoch) = self.epoch {
            f::native_store::encode_native_paged_ownership_epoch(&mut data, epoch)?;
        }
        if let Some(proofs) = proofs {
            f::page_proof::encode_catalog(&mut data, proofs)?;
        }
        if let Some(mut resume) = self.resume.clone() {
            for state in resume.collections.values_mut() {
                state.root_page = remap(state.root_page)?;
                state.key_index_root = state.key_index_root.map(remap).transpose()?;
                for index in state.secondary_indices.values_mut() {
                    index.root_page = remap(index.root_page)?;
                }
            }
            f::native_resume::encode_catalog_resume(&mut data, &resume)?;
        }
        Ok(data)
    }
}

pub(super) fn write_catalog(target: &Pager, data: &[u8]) -> Result<()> {
    crate::storage::unified::UnifiedStore::publish_native_catalog(target, data)?;
    Ok(())
}

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

    fn resume(root_page: u32) -> f::native_resume::CollectionResume {
        f::native_resume::CollectionResume {
            root_page,
            entities: 1,
            entity_id_next: 2,
            row_id_next: 2,
            sequence_id_next: 2,
            key_index_root: None,
            secondary_indices: BTreeMap::new(),
            metadata_expiries: Vec::new(),
        }
    }

    fn duplicate_config_catalog() -> Catalog {
        Catalog {
            proofs: None,
            header: f::NativePagedMetadataHeader {
                format_version: 11,
                collection_count: 3,
            },
            roots: vec![
                f::NativePagedCollectionRoot {
                    collection: "users".into(),
                    root_page: 10,
                },
                f::NativePagedCollectionRoot {
                    collection: "red_config".into(),
                    root_page: 20,
                },
                f::NativePagedCollectionRoot {
                    collection: "red.config".into(),
                    root_page: 30,
                },
            ],
            references: Vec::new(),
            epoch: Some(7),
            retired_roots: Vec::new(),
            deferred_pages: Vec::new(),
            resume: Some(f::native_resume::CatalogResume {
                entity_id_next: 100,
                xid_high_water_mark: Some(8),
                metadata_expiry_prepared: false,
                collections: [
                    ("users".into(), resume(10)),
                    ("red_config".into(), resume(20)),
                    ("red.config".into(), resume(30)),
                ]
                .into_iter()
                .collect(),
            }),
        }
    }

    #[test]
    fn explicit_repair_retires_only_legacy_config_identity() {
        let mut catalog = duplicate_config_catalog();
        catalog.retain_canonical_system_config().unwrap();
        assert_eq!(catalog.header.collection_count, 2);
        assert_eq!(
            catalog
                .roots
                .iter()
                .map(|root| root.collection.as_str())
                .collect::<Vec<_>>(),
            ["users", "red.config"]
        );
        let resume = catalog.resume.unwrap();
        assert!(!resume.collections.contains_key("red_config"));
        assert_eq!(resume.collections["red.config"].root_page, 30);
        assert_eq!(resume.collections["users"].root_page, 10);
    }

    #[test]
    fn explicit_repair_rejects_legacy_cross_references() {
        let mut catalog = duplicate_config_catalog();
        catalog.references.push(f::NativePagedCrossRef {
            source_id: 1,
            target_id: 2,
            ref_type: 0,
            target_collection: "red_config".into(),
        });
        assert!(catalog
            .retain_canonical_system_config()
            .unwrap_err()
            .to_string()
            .contains("cross-references"));
    }

    #[test]
    fn migration_promotes_populated_legacy_config_and_retires_empty_canonical() {
        let mut catalog = duplicate_config_catalog();
        catalog
            .resume
            .as_mut()
            .unwrap()
            .collections
            .get_mut("red.config")
            .unwrap()
            .entities = 0;

        assert_eq!(catalog.migrate_legacy_system_config().unwrap(), 1);
        assert_eq!(catalog.header.collection_count, 2);
        assert_eq!(
            catalog
                .roots
                .iter()
                .map(|root| (root.collection.as_str(), root.root_page))
                .collect::<Vec<_>>(),
            [("users", 10), ("red.config", 20)]
        );
        let resume = catalog.resume.unwrap();
        assert!(!resume.collections.contains_key("red_config"));
        assert_eq!(resume.collections["red.config"].root_page, 20);
    }

    #[test]
    fn migration_promotes_single_legacy_config_identity() {
        let mut catalog = duplicate_config_catalog();
        catalog.roots.retain(|root| root.collection != "red.config");
        catalog.header.collection_count = 2;
        catalog
            .resume
            .as_mut()
            .unwrap()
            .collections
            .remove("red.config");

        assert_eq!(catalog.migrate_legacy_system_config().unwrap(), 1);
        assert_eq!(catalog.header.collection_count, 2);
        assert_eq!(
            catalog
                .roots
                .iter()
                .map(|root| (root.collection.as_str(), root.root_page))
                .collect::<Vec<_>>(),
            [("users", 10), ("red.config", 20)]
        );
        let resume = catalog.resume.unwrap();
        assert!(!resume.collections.contains_key("red_config"));
        assert_eq!(resume.collections["red.config"].root_page, 20);
    }

    #[test]
    fn migration_accepts_single_canonical_config_identity() {
        let mut catalog = duplicate_config_catalog();
        catalog.roots.retain(|root| root.collection != "red_config");
        catalog.header.collection_count = 2;
        catalog
            .resume
            .as_mut()
            .unwrap()
            .collections
            .remove("red_config");

        assert_eq!(catalog.migrate_legacy_system_config().unwrap(), 0);
        assert_eq!(catalog.header.collection_count, 2);
        assert_eq!(
            catalog
                .roots
                .iter()
                .map(|root| (root.collection.as_str(), root.root_page))
                .collect::<Vec<_>>(),
            [("users", 10), ("red.config", 30)]
        );
        let resume = catalog.resume.unwrap();
        assert!(!resume.collections.contains_key("red_config"));
        assert_eq!(resume.collections["red.config"].root_page, 30);
    }

    #[test]
    fn migration_refuses_catalog_without_any_config_identity() {
        let mut catalog = duplicate_config_catalog();
        catalog
            .roots
            .retain(|root| root.collection != "red_config" && root.collection != "red.config");
        catalog.header.collection_count = 1;
        let resume = catalog.resume.as_mut().unwrap();
        resume.collections.remove("red_config");
        resume.collections.remove("red.config");

        assert!(catalog
            .migrate_legacy_system_config()
            .unwrap_err()
            .to_string()
            .contains("requires a config root"));
    }

    #[test]
    fn migration_refuses_two_populated_config_trees() {
        let mut catalog = duplicate_config_catalog();
        assert!(catalog
            .migrate_legacy_system_config()
            .unwrap_err()
            .to_string()
            .contains("refusing lossy migration"));
    }
}