Skip to main content

reddb_server/storage/unified/devx/reddb/
impl_registry.rs

1use super::*;
2
3impl RedDB {
4    pub(crate) fn write_column_block_page(
5        &self,
6        bytes: &[u8],
7    ) -> Result<crate::storage::engine::PageLocation, Box<dyn std::error::Error>> {
8        use crate::storage::engine::PageType;
9
10        let store = self.store();
11        let Some(pager) = store.pager() else {
12            return Ok(crate::storage::engine::PageLocation::new(
13                0,
14                0,
15                bytes.len() as u32,
16            ));
17        };
18        let content_size = crate::storage::engine::page::CONTENT_SIZE;
19        let page_count = bytes.len().max(1).div_ceil(content_size);
20        let mut first_page_id = None;
21        for page_index in 0..page_count {
22            let mut page = pager.allocate_page(PageType::ColumnBlock)?;
23            let page_id = page.page_id();
24            if let Some(first) = first_page_id {
25                let expected = first + page_index as u32;
26                if page_id != expected {
27                    return Err(format!(
28                        "column block page allocation was not contiguous: got {page_id}, expected {expected}"
29                    )
30                    .into());
31                }
32            } else {
33                first_page_id = Some(page_id);
34            }
35            let start = page_index * content_size;
36            let end = (start + content_size).min(bytes.len());
37            if start < end {
38                page.content_mut()[..end - start].copy_from_slice(&bytes[start..end]);
39            }
40            pager.write_page_encrypted(page_id, page)?;
41        }
42        Ok(crate::storage::engine::PageLocation::new(
43            first_page_id.expect("page_count is at least one"),
44            0,
45            bytes.len() as u32,
46        ))
47    }
48
49    pub(crate) fn read_column_block_page(
50        &self,
51        loc: crate::storage::engine::PageLocation,
52    ) -> Result<Option<Vec<u8>>, Box<dyn std::error::Error>> {
53        let store = self.store();
54        let Some(pager) = store.pager() else {
55            return Ok(None);
56        };
57        let content_size = crate::storage::engine::page::CONTENT_SIZE;
58        let mut remaining = loc.length as usize;
59        let mut page_id = loc.page_id;
60        let mut offset = loc.offset as usize;
61        let mut out = Vec::with_capacity(remaining);
62        while remaining > 0 {
63            let page = pager.read_page_decrypted(page_id)?;
64            if page.page_type()? != crate::storage::engine::PageType::ColumnBlock {
65                return Err(format!("page {page_id} is not a ColumnBlock page").into());
66            }
67            if offset > content_size {
68                return Err(format!(
69                    "column block page location offset out of bounds: {offset} > {content_size}"
70                )
71                .into());
72            }
73            let available = content_size - offset;
74            let take = remaining.min(available);
75            out.extend_from_slice(&page.content()[offset..offset + take]);
76            remaining -= take;
77            page_id = page_id.saturating_add(1);
78            offset = 0;
79        }
80        Ok(Some(out))
81    }
82
83    pub fn tree_definitions(&self) -> Vec<crate::physical::PhysicalTreeDefinition> {
84        self.physical_metadata()
85            .map(|metadata| metadata.tree_definitions)
86            .unwrap_or_default()
87    }
88
89    pub fn tree_definition(
90        &self,
91        collection: &str,
92        name: &str,
93    ) -> Option<crate::physical::PhysicalTreeDefinition> {
94        self.tree_definitions()
95            .into_iter()
96            .find(|definition| definition.collection == collection && definition.name == name)
97    }
98
99    pub fn save_tree_definition(
100        &self,
101        definition: crate::physical::PhysicalTreeDefinition,
102    ) -> Result<crate::physical::PhysicalTreeDefinition, Box<dyn std::error::Error>> {
103        self.update_physical_metadata(|metadata| {
104            if let Some(existing) = metadata.tree_definitions.iter_mut().find(|existing| {
105                existing.collection == definition.collection && existing.name == definition.name
106            }) {
107                *existing = definition.clone();
108            } else {
109                metadata.tree_definitions.push(definition.clone());
110            }
111            metadata.tree_definitions.sort_by(|left, right| {
112                left.collection
113                    .cmp(&right.collection)
114                    .then_with(|| left.name.cmp(&right.name))
115            });
116            definition.clone()
117        })
118    }
119
120    pub fn remove_tree_definition(
121        &self,
122        collection: &str,
123        name: &str,
124    ) -> Result<Option<crate::physical::PhysicalTreeDefinition>, Box<dyn std::error::Error>> {
125        self.update_physical_metadata(|metadata| {
126            metadata
127                .tree_definitions
128                .iter()
129                .position(|definition| {
130                    definition.collection == collection && definition.name == name
131                })
132                .map(|index| metadata.tree_definitions.remove(index))
133        })
134    }
135
136    pub fn collection_default_ttl_ms(&self, collection: &str) -> Option<u64> {
137        self.collection_ttl_defaults_ms
138            .read()
139            .ok()
140            .and_then(|defaults| defaults.get(collection).copied())
141    }
142
143    pub fn set_collection_default_ttl_ms(&self, collection: impl Into<String>, ttl_ms: u64) {
144        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
145            defaults.insert(collection.into(), ttl_ms);
146        }
147    }
148
149    pub fn clear_collection_default_ttl_ms(&self, collection: &str) {
150        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
151            defaults.remove(collection);
152        }
153    }
154
155    pub fn collection_contracts(&self) -> Vec<crate::physical::CollectionContract> {
156        self.contract_cache_map()
157            .values()
158            .map(|arc| (**arc).clone())
159            .collect()
160    }
161
162    pub fn collection_contract(
163        &self,
164        collection: &str,
165    ) -> Option<crate::physical::CollectionContract> {
166        self.contract_cache_map()
167            .get(collection)
168            .map(|arc| (**arc).clone())
169    }
170
171    /// Arc-bumping variant — callers that only need read access skip
172    /// the per-call full `CollectionContract` clone. Deep clones cost
173    /// ~200-400 ns each (vec of declared columns + string allocs);
174    /// the UPDATE fast path hits this 2-3× per mutation.
175    pub fn collection_contract_arc(
176        &self,
177        collection: &str,
178    ) -> Option<std::sync::Arc<crate::physical::CollectionContract>> {
179        self.contract_cache_map()
180            .get(collection)
181            .map(std::sync::Arc::clone)
182    }
183
184    /// Return (or lazily build) the in-memory contract map.
185    /// Reparses PhysicalMetadataFile JSON only once per invalidation.
186    /// Called 3× per insert (ensure_model, enforce_uniqueness, normalize_fields)
187    /// so the cache is load-bearing for insert throughput.
188    fn contract_cache_map(
189        &self,
190    ) -> std::sync::Arc<
191        std::collections::HashMap<String, std::sync::Arc<crate::physical::CollectionContract>>,
192    > {
193        self.collection_contract_cache.get_or_load(|| {
194            self.physical_metadata()
195                .map(|metadata| metadata.collection_contracts)
196                .unwrap_or_default()
197        })
198    }
199
200    pub(crate) fn invalidate_collection_contract_cache(&self) {
201        self.collection_contract_cache.invalidate();
202    }
203
204    /// Single-file artifact only: mirror the current contract set into the
205    /// store's auxiliary-metadata blob so it rides the binary dump. Without
206    /// this the contracts live only in the in-memory cache and a collection's
207    /// declared model (e.g. `kv`) is lost on reopen — recovery would re-infer
208    /// it from the stored entities and a KV collection would come back as a
209    /// table. No-op for other profiles, which persist contracts via the
210    /// metadata sidecar.
211    pub(crate) fn sync_single_file_contract_blob(&self) -> Result<(), Box<dyn std::error::Error>> {
212        let contracts = self.collection_contracts();
213        let bytes = crate::physical::serialize_collection_contracts(&contracts);
214        self.store().persist_aux_metadata(bytes)?;
215        Ok(())
216    }
217
218    /// Seed the in-memory contract cache from the store's auxiliary-metadata
219    /// blob (single-file artifact). Called on open so a collection's declared
220    /// model survives a restart. Does nothing when the blob is absent (older
221    /// V9 dumps) or empty.
222    pub(crate) fn seed_contract_cache_from_store_aux(&self) -> std::io::Result<()> {
223        let bytes = self.store().aux_metadata();
224        if bytes.is_empty() {
225            return Ok(());
226        }
227        let contracts = crate::physical::deserialize_collection_contracts(&bytes)?;
228        let map: std::collections::HashMap<_, _> = contracts
229            .into_iter()
230            .map(|contract| (contract.name.clone(), std::sync::Arc::new(contract)))
231            .collect();
232        // Restore the per-collection state derived from contracts (context
233        // index enablement + default TTLs) so the full collection definition —
234        // not just the declared model — survives a restart.
235        let store = self.store();
236        let index = store.context_index();
237        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
238            for contract in map.values() {
239                index.set_collection_enabled(&contract.name, contract.context_index_enabled);
240                if let Some(ttl_ms) = contract.default_ttl_ms {
241                    defaults.insert(contract.name.clone(), ttl_ms);
242                }
243            }
244        }
245        self.collection_contract_cache.replace(map);
246        Ok(())
247    }
248
249    fn update_collection_contract_metadata<T, F>(
250        &self,
251        mutator: F,
252    ) -> Result<T, Box<dyn std::error::Error>>
253    where
254        F: FnOnce(&mut PhysicalMetadataFile) -> T,
255    {
256        if !self.paged_mode
257            || self.options.storage_profile.deploy_profile
258                != crate::storage::DeployProfile::Embedded
259            || self.options.storage_profile.packaging
260                != crate::storage::StoragePackaging::SingleFile
261        {
262            return self.update_physical_metadata(mutator);
263        }
264        if self.options.read_only {
265            return Err("physical metadata operations are not allowed in read-only mode".into());
266        }
267        let Some(path) = self.path() else {
268            return Err("database path is not available".into());
269        };
270
271        // Legacy paged files keep contracts in the durable metadata sidecar,
272        // not the embedded dump's auxiliary blob or native artifact summaries.
273        // A definition edit must not rebuild every collection's search indexes.
274        // Retain publication serialization and the canonical atomic writer.
275        let _publication = self.store.vault_publication_lock.lock();
276        let mut metadata = self.load_or_bootstrap_physical_metadata(false)?;
277        let result = mutator(&mut metadata);
278        metadata.save_for_data_path(path)?;
279        Ok(result)
280    }
281
282    pub fn save_collection_contract(
283        &self,
284        contract: crate::physical::CollectionContract,
285    ) -> Result<crate::physical::CollectionContract, Box<dyn std::error::Error>> {
286        let _publication = self.store.vault_publication_lock.lock();
287        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
288            if let Some(ttl_ms) = contract.default_ttl_ms {
289                defaults.insert(contract.name.clone(), ttl_ms);
290            } else {
291                defaults.remove(&contract.name);
292            }
293        }
294
295        self.store()
296            .context_index()
297            .set_collection_enabled(&contract.name, contract.context_index_enabled);
298        if let Some(manager) = self.store().get_collection(&contract.name) {
299            let columns = contract
300                .declared_columns
301                .iter()
302                .map(|column| column.name.clone())
303                .collect();
304            manager.set_column_schema_if_empty(columns);
305        }
306
307        // Existing paged files can be opened with embedded options. Their
308        // checkpoints do not encode the binary dump's auxiliary contract blob.
309        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
310            && self.options.storage_profile.packaging
311                == crate::storage::StoragePackaging::SingleFile
312            && !self.paged_mode
313        {
314            self.collection_contract_cache.insert(contract.clone());
315            self.sync_single_file_contract_blob()?;
316            return Ok(contract);
317        }
318
319        self.invalidate_collection_contract_cache();
320
321        let saved = self.update_collection_contract_metadata(|metadata| {
322            if let Some(existing) = metadata
323                .collection_contracts
324                .iter_mut()
325                .find(|existing| existing.name == contract.name)
326            {
327                *existing = contract.clone();
328            } else {
329                metadata.collection_contracts.push(contract.clone());
330            }
331            metadata
332                .collection_contracts
333                .sort_by(|left, right| left.name.cmp(&right.name));
334
335            if let Some(ttl_ms) = contract.default_ttl_ms {
336                metadata
337                    .collection_ttl_defaults_ms
338                    .insert(contract.name.clone(), ttl_ms);
339            } else {
340                metadata.collection_ttl_defaults_ms.remove(&contract.name);
341            }
342
343            contract.clone()
344        })?;
345        self.invalidate_collection_contract_cache();
346        Ok(saved)
347    }
348
349    pub fn remove_collection_contract(
350        &self,
351        collection: &str,
352    ) -> Result<Option<crate::physical::CollectionContract>, Box<dyn std::error::Error>> {
353        let _publication = self.store.vault_publication_lock.lock();
354        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
355            defaults.remove(collection);
356        }
357
358        self.store()
359            .context_index()
360            .set_collection_enabled(collection, false);
361
362        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
363            && self.options.storage_profile.packaging
364                == crate::storage::StoragePackaging::SingleFile
365            && !self.paged_mode
366        {
367            let removed = self.collection_contract_cache.remove(collection);
368            self.sync_single_file_contract_blob()?;
369            return Ok(removed);
370        }
371
372        self.invalidate_collection_contract_cache();
373
374        if self.paged_mode {
375            let Some(path) = self.path() else {
376                return Err("database path is not available".into());
377            };
378            // DROP already published the native collection registry before it
379            // reaches this contract edit.  Rebuild only the small physical
380            // metadata catalogue from its last durable roots, removing the
381            // dropped root and descriptors explicitly.  Calling the generic
382            // metadata updater here used to fingerprint every row in changed
383            // internal collections (notably append-heavy red_config) while
384            // holding the publication lock, turning a tiny DROP after cold
385            // boot into work proportional to the database's history.
386            //
387            // Other collection fingerprints describe their last checkpoint,
388            // just as they do between ordinary DML and the next checkpoint;
389            // the canonical checkpoint path remains responsible for
390            // recomputing changed fingerprints.
391            let previous = self.load_or_bootstrap_physical_metadata(false)?;
392            let mut roots = previous.superblock.collection_roots.clone();
393            roots.remove(collection);
394            let indexes = previous
395                .indexes
396                .iter()
397                .filter(|index| index.collection.as_deref() != Some(collection))
398                .cloned()
399                .collect();
400            let mut metadata = PhysicalMetadataFile::from_state(
401                self.options.clone(),
402                self.catalog_snapshot(),
403                roots,
404                indexes,
405                Some(&previous),
406            );
407            let removed = metadata
408                .collection_contracts
409                .iter()
410                .position(|contract| contract.name == collection)
411                .map(|index| metadata.collection_contracts.remove(index));
412            metadata.collection_ttl_defaults_ms.remove(collection);
413            metadata.save_for_data_path(path)?;
414            self.persist_native_physical_header(&metadata)?;
415            self.invalidate_collection_contract_cache();
416            return Ok(removed);
417        }
418
419        let removed = self.update_collection_contract_metadata(|metadata| {
420            let removed = metadata
421                .collection_contracts
422                .iter()
423                .position(|contract| contract.name == collection)
424                .map(|index| metadata.collection_contracts.remove(index));
425            metadata.collection_ttl_defaults_ms.remove(collection);
426            metadata
427                .indexes
428                .retain(|index| index.collection.as_deref() != Some(collection));
429            removed
430        })?;
431        self.invalidate_collection_contract_cache();
432        Ok(removed)
433    }
434
435    pub(crate) fn collection_ttl_defaults_snapshot(&self) -> BTreeMap<String, u64> {
436        self.collection_ttl_defaults_ms
437            .read()
438            .map(|defaults| {
439                defaults
440                    .iter()
441                    .map(|(collection, ttl_ms)| (collection.clone(), *ttl_ms))
442                    .collect()
443            })
444            .unwrap_or_default()
445    }
446
447    pub(crate) fn load_collection_ttl_defaults_from_metadata(&self) {
448        let metadata = self.physical_metadata();
449        let defaults = metadata
450            .as_ref()
451            .map(|m| m.collection_ttl_defaults_ms.clone())
452            .unwrap_or_default();
453
454        if let Ok(mut current) = self.collection_ttl_defaults_ms.write() {
455            current.clear();
456            current.extend(defaults);
457        }
458
459        if let Some(metadata) = metadata {
460            let store = self.store();
461            let index = store.context_index();
462            for contract in &metadata.collection_contracts {
463                index.set_collection_enabled(&contract.name, contract.context_index_enabled);
464            }
465        }
466    }
467
468    /// Snapshot the live hypertable registry (specs + every chunk)
469    /// into the sidecar-serialisable form. Called from
470    /// [`persist_metadata`] so the chunk routing spine is durable on
471    /// the same metadata path as collection contracts — issue #866.
472    /// Returns an empty vec (and touches nothing) when no hypertable
473    /// has ever been declared, so non-hypertable databases pay no
474    /// persist cost.
475    pub(crate) fn hypertable_registry_snapshot(&self) -> Vec<crate::physical::PhysicalHypertable> {
476        let registry = self.hypertables();
477        if registry.is_empty() {
478            return Vec::new();
479        }
480        use std::collections::BTreeMap;
481        let mut chunks_by_table: BTreeMap<String, Vec<crate::physical::PhysicalHypertableChunk>> =
482            BTreeMap::new();
483        for chunk in registry.snapshot_chunks() {
484            chunks_by_table
485                .entry(chunk.id.hypertable.clone())
486                .or_default()
487                .push(crate::physical::PhysicalHypertableChunk {
488                    start_ns: chunk.id.start_ns,
489                    end_ns_exclusive: chunk.end_ns_exclusive,
490                    row_count: chunk.row_count,
491                    min_ts_ns: chunk.min_ts_ns,
492                    max_ts_ns: chunk.max_ts_ns,
493                    sealed: chunk.sealed,
494                    ttl_override_ns: chunk.ttl_override_ns,
495                    columnar_page: chunk.columnar_page,
496                    columnar_derived: chunk.columnar_page.is_some(),
497                });
498        }
499        registry
500            .list()
501            .into_iter()
502            .map(|spec| crate::physical::PhysicalHypertable {
503                chunks: chunks_by_table.remove(&spec.name).unwrap_or_default(),
504                name: spec.name,
505                time_column: spec.time_column,
506                chunk_interval_ns: spec.chunk_interval_ns,
507                default_ttl_ns: spec.default_ttl_ns,
508            })
509            .collect()
510    }
511
512    /// Rehydrate the hypertable registry from the physical metadata
513    /// sidecar at boot. Mirror of
514    /// [`load_collection_ttl_defaults_from_metadata`] — specs and
515    /// chunks are restored verbatim so bounds / routing / TTL are
516    /// identical to the pre-restart state (issue #866).
517    pub(crate) fn load_hypertables_from_metadata(&self) {
518        let Some(metadata) = self.physical_metadata() else {
519            return;
520        };
521        if metadata.hypertables.is_empty() {
522            return;
523        }
524        let registry = self.hypertables();
525        for hypertable in &metadata.hypertables {
526            let mut spec = crate::storage::timeseries::HypertableSpec::new(
527                hypertable.name.clone(),
528                hypertable.time_column.clone(),
529                hypertable.chunk_interval_ns,
530            );
531            if let Some(ttl) = hypertable.default_ttl_ns {
532                spec = spec.with_ttl_ns(ttl);
533            }
534            registry.register(spec);
535            for chunk in &hypertable.chunks {
536                let meta = crate::storage::timeseries::ChunkMeta {
537                    id: crate::storage::timeseries::ChunkId {
538                        hypertable: hypertable.name.clone(),
539                        start_ns: chunk.start_ns,
540                    },
541                    end_ns_exclusive: chunk.end_ns_exclusive,
542                    row_count: chunk.row_count,
543                    min_ts_ns: chunk.min_ts_ns,
544                    max_ts_ns: chunk.max_ts_ns,
545                    sealed: chunk.sealed,
546                    ttl_override_ns: chunk.ttl_override_ns,
547                    columnar_page: chunk.columnar_page,
548                };
549                let id = meta.id.clone();
550                let columnar_page = meta.columnar_page;
551                registry.restore_chunk(meta);
552                if let Some(loc) = columnar_page {
553                    if let Ok(Some(bytes)) = self.read_column_block_page(loc) {
554                        registry.restore_columnar_block(&id, bytes);
555                    }
556                }
557            }
558        }
559    }
560
561    pub fn run_maintenance(&self) -> Result<(), Box<dyn std::error::Error>> {
562        self.store.run_maintenance()?;
563        self.persist_metadata()?;
564        Ok(())
565    }
566
567    /// Path to the physical metadata sidecar, if persistent.
568    pub fn metadata_path(&self) -> Option<PathBuf> {
569        self.path
570            .as_ref()
571            .map(|path| PhysicalMetadataFile::metadata_path_for(path))
572    }
573
574    /// Inspect the current physical metadata view without publishing repairs.
575    pub fn physical_metadata(&self) -> Option<PhysicalMetadataFile> {
576        self.load_or_bootstrap_physical_metadata(false).ok()
577    }
578
579    /// Physical index registry derived for the current database state.
580    pub fn physical_indexes(&self) -> Vec<PhysicalIndexState> {
581        let indexes = self
582            .physical_metadata()
583            .map(|metadata| metadata.indexes)
584            .unwrap_or_default();
585        self.physical_indexes_from_declarations(indexes)
586    }
587
588    pub(crate) fn physical_indexes_from_declarations(
589        &self,
590        indexes: Vec<PhysicalIndexState>,
591    ) -> Vec<PhysicalIndexState> {
592        let indexes = Some(indexes)
593            .filter(|indexes| !indexes.is_empty())
594            .or_else(|| {
595                self.native_physical_state()
596                    .map(|state| self.physical_index_state_from_native_state(&state, None))
597            })
598            .unwrap_or_else(|| self.physical_index_state());
599        self.reconcile_index_states_with_native_artifacts(indexes)
600    }
601
602    /// List registered named exports from the current physical metadata view.
603    pub fn exports(&self) -> Vec<ExportDescriptor> {
604        self.physical_metadata()
605            .map(|metadata| metadata.exports)
606            .or_else(|| {
607                self.native_physical_state()
608                    .map(|state| self.exports_from_native_state(&state))
609            })
610            .unwrap_or_default()
611    }
612
613    /// List recorded snapshots from the current physical metadata view.
614    pub fn snapshots(&self) -> Vec<crate::physical::SnapshotDescriptor> {
615        if let Some(metadata) = self.physical_metadata() {
616            if !metadata.snapshots.is_empty() {
617                return metadata.snapshots;
618            }
619        }
620        self.native_physical_state()
621            .map(|state| self.snapshots_from_native_state(&state))
622            .unwrap_or_default()
623    }
624
625    /// List persisted named graph projections from the current physical metadata view.
626    pub fn graph_projections(&self) -> Vec<PhysicalGraphProjection> {
627        self.physical_metadata()
628            .map(|metadata| metadata.graph_projections)
629            .or_else(|| {
630                self.native_physical_state()
631                    .map(|state| self.graph_projections_from_native_state(&state))
632            })
633            .unwrap_or_default()
634    }
635
636    /// List graph projections declared in the catalog view.
637    pub fn declared_graph_projections(&self) -> Vec<PhysicalGraphProjection> {
638        self.catalog_model_snapshot().declared_graph_projections
639    }
640
641    /// List graph projections currently observed in the operational view.
642    pub fn operational_graph_projections(&self) -> Vec<PhysicalGraphProjection> {
643        self.graph_projections()
644            .into_iter()
645            .filter(|projection| {
646                projection.last_materialized_sequence.is_some()
647                    || matches!(projection.state.as_str(), "materialized" | "stale")
648            })
649            .collect()
650    }
651
652    /// List persisted analytics job metadata from the current physical metadata view.
653    pub fn analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
654        self.physical_metadata()
655            .map(|metadata| metadata.analytics_jobs)
656            .or_else(|| {
657                self.native_physical_state()
658                    .map(|state| self.analytics_jobs_from_native_state(&state))
659            })
660            .unwrap_or_default()
661    }
662
663    /// List analytics jobs declared in the catalog view.
664    pub fn declared_analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
665        self.catalog_model_snapshot().declared_analytics_jobs
666    }
667
668    /// List analytics jobs currently observed in the operational view.
669    pub fn operational_analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
670        self.analytics_jobs()
671            .into_iter()
672            .filter(|job| {
673                job.last_run_sequence.is_some()
674                    || matches!(
675                        job.state.as_str(),
676                        "running" | "completed" | "failed" | "queued" | "stale"
677                    )
678            })
679            .collect()
680    }
681
682    /// List indexes declared in the catalog view.
683    pub fn declared_indexes(&self) -> Vec<PhysicalIndexState> {
684        self.catalog_model_snapshot().declared_indexes
685    }
686
687    /// List indexes currently observed in the operational view.
688    pub fn operational_indexes(&self) -> Vec<PhysicalIndexState> {
689        self.catalog_model_snapshot().operational_indexes
690    }
691
692    /// List reconciled index status entries from the catalog snapshot.
693    pub fn index_statuses(&self) -> Vec<CatalogIndexStatus> {
694        self.catalog_model_snapshot().index_statuses
695    }
696
697    /// Resolve one index status entry from the catalog snapshot.
698    pub fn index_status(&self, name: &str) -> Option<CatalogIndexStatus> {
699        self.catalog_model_snapshot()
700            .index_statuses
701            .into_iter()
702            .find(|status| status.name == name)
703    }
704
705    /// Upsert a named graph projection in the persisted physical metadata.
706    pub fn save_graph_projection(
707        &self,
708        name: impl Into<String>,
709        node_labels: Vec<String>,
710        node_types: Vec<String>,
711        edge_labels: Vec<String>,
712        source: impl Into<String>,
713    ) -> Result<PhysicalGraphProjection, Box<dyn std::error::Error>> {
714        let name = name.into();
715        let source = source.into();
716        self.update_physical_metadata(|metadata| {
717            let now = SystemTime::now()
718                .duration_since(UNIX_EPOCH)
719                .unwrap_or_default()
720                .as_millis();
721            let projection = if let Some(existing) = metadata
722                .graph_projections
723                .iter_mut()
724                .find(|projection| projection.name == name)
725            {
726                existing.updated_at_unix_ms = now;
727                existing.state = "declared".to_string();
728                existing.source = source.clone();
729                existing.node_labels = node_labels.clone();
730                existing.node_types = node_types.clone();
731                existing.edge_labels = edge_labels.clone();
732                existing.last_materialized_sequence = None;
733                existing.clone()
734            } else {
735                let projection = PhysicalGraphProjection {
736                    name: name.clone(),
737                    created_at_unix_ms: now,
738                    updated_at_unix_ms: now,
739                    state: "declared".to_string(),
740                    source: source.clone(),
741                    node_labels: node_labels.clone(),
742                    node_types: node_types.clone(),
743                    edge_labels: edge_labels.clone(),
744                    last_materialized_sequence: None,
745                };
746                metadata.graph_projections.push(projection.clone());
747                projection
748            };
749
750            Self::mark_projection_dependent_jobs_stale(metadata, &name, now);
751
752            metadata
753                .graph_projections
754                .sort_by(|left, right| left.name.cmp(&right.name));
755            projection
756        })
757    }
758
759    /// Mark a declared graph projection as materialized in the current physical metadata.
760    pub fn materialize_graph_projection(
761        &self,
762        name: &str,
763    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
764        self.update_physical_metadata(|metadata| {
765            let now = SystemTime::now()
766                .duration_since(UNIX_EPOCH)
767                .unwrap_or_default()
768                .as_millis();
769            let idx = metadata
770                .graph_projections
771                .iter()
772                .position(|projection| projection.name == name);
773            if let Some(idx) = idx {
774                metadata.graph_projections[idx].updated_at_unix_ms = now;
775                metadata.graph_projections[idx].state = "materialized".to_string();
776                metadata.graph_projections[idx].last_materialized_sequence =
777                    Some(metadata.superblock.sequence);
778                let result = metadata.graph_projections[idx].clone();
779                Self::rearm_projection_dependent_jobs_declared(metadata, name, now);
780                return Some(result);
781            }
782            None
783        })
784    }
785
786    /// Mark a declared graph projection as materializing.
787    pub fn mark_graph_projection_materializing(
788        &self,
789        name: &str,
790    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
791        self.update_physical_metadata(|metadata| {
792            let now = SystemTime::now()
793                .duration_since(UNIX_EPOCH)
794                .unwrap_or_default()
795                .as_millis();
796            let idx = metadata
797                .graph_projections
798                .iter()
799                .position(|projection| projection.name == name);
800            if let Some(idx) = idx {
801                metadata.graph_projections[idx].updated_at_unix_ms = now;
802                metadata.graph_projections[idx].state = "materializing".to_string();
803                metadata.graph_projections[idx].last_materialized_sequence = None;
804                let result = metadata.graph_projections[idx].clone();
805                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
806                return Some(result);
807            }
808            None
809        })
810    }
811
812    /// Mark a graph projection as failed.
813    pub fn fail_graph_projection(
814        &self,
815        name: &str,
816    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
817        self.update_physical_metadata(|metadata| {
818            let now = SystemTime::now()
819                .duration_since(UNIX_EPOCH)
820                .unwrap_or_default()
821                .as_millis();
822            let idx = metadata
823                .graph_projections
824                .iter()
825                .position(|projection| projection.name == name);
826            if let Some(idx) = idx {
827                metadata.graph_projections[idx].updated_at_unix_ms = now;
828                metadata.graph_projections[idx].state = "failed".to_string();
829                metadata.graph_projections[idx].last_materialized_sequence = None;
830                let result = metadata.graph_projections[idx].clone();
831                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
832                return Some(result);
833            }
834            None
835        })
836    }
837
838    /// Mark a graph projection as stale while preserving any last materialized sequence.
839    pub fn mark_graph_projection_stale(
840        &self,
841        name: &str,
842    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
843        self.update_physical_metadata(|metadata| {
844            let now = SystemTime::now()
845                .duration_since(UNIX_EPOCH)
846                .unwrap_or_default()
847                .as_millis();
848            let idx = metadata
849                .graph_projections
850                .iter()
851                .position(|projection| projection.name == name);
852            if let Some(idx) = idx {
853                metadata.graph_projections[idx].updated_at_unix_ms = now;
854                metadata.graph_projections[idx].state = "stale".to_string();
855                let result = metadata.graph_projections[idx].clone();
856                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
857                return Some(result);
858            }
859            None
860        })
861    }
862
863    fn mark_projection_dependent_jobs_stale(
864        metadata: &mut PhysicalMetadataFile,
865        projection_name: &str,
866        now: u128,
867    ) {
868        for job in metadata.analytics_jobs.iter_mut() {
869            if job.projection.as_deref() == Some(projection_name) && job.state != "declared" {
870                job.state = "stale".to_string();
871                job.updated_at_unix_ms = now;
872            }
873        }
874    }
875
876    fn rearm_projection_dependent_jobs_declared(
877        metadata: &mut PhysicalMetadataFile,
878        projection_name: &str,
879        now: u128,
880    ) {
881        for job in metadata.analytics_jobs.iter_mut() {
882            if job.projection.as_deref() == Some(projection_name) && job.state == "stale" {
883                job.state = "declared".to_string();
884                job.last_run_sequence = None;
885                job.updated_at_unix_ms = now;
886            }
887        }
888    }
889
890    /// Declare or update analytics job metadata without marking it as executed.
891    pub fn save_analytics_job(
892        &self,
893        kind: impl Into<String>,
894        projection: Option<String>,
895        metadata_entries: BTreeMap<String, String>,
896    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
897        let kind = kind.into();
898        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
899        if let Some(projection_name) = projection.as_deref() {
900            if !self.graph_projection_is_declared(projection_name) {
901                return Err(format!(
902                    "graph projection '{projection_name}' is not declared in physical metadata"
903                )
904                .into());
905            }
906        }
907
908        self.update_physical_metadata(|metadata| {
909            let now = SystemTime::now()
910                .duration_since(UNIX_EPOCH)
911                .unwrap_or_default()
912                .as_millis();
913
914            let job = if let Some(existing) = metadata
915                .analytics_jobs
916                .iter_mut()
917                .find(|job| job.id == job_id)
918            {
919                existing.kind = kind.clone();
920                existing.projection = projection.clone();
921                existing.updated_at_unix_ms = now;
922                existing.metadata = metadata_entries.clone();
923                if existing.last_run_sequence.is_none() {
924                    existing.state = "declared".to_string();
925                }
926                existing.clone()
927            } else {
928                let job = PhysicalAnalyticsJob {
929                    id: job_id.clone(),
930                    kind: kind.clone(),
931                    state: "declared".to_string(),
932                    projection: projection.clone(),
933                    created_at_unix_ms: now,
934                    updated_at_unix_ms: now,
935                    last_run_sequence: None,
936                    metadata: metadata_entries.clone(),
937                };
938                metadata.analytics_jobs.push(job.clone());
939                job
940            };
941
942            metadata
943                .analytics_jobs
944                .sort_by(|left, right| left.id.cmp(&right.id));
945            job
946        })
947    }
948
949    /// Record or update analytics job metadata in the persisted physical metadata.
950    pub fn record_analytics_job(
951        &self,
952        kind: impl Into<String>,
953        projection: Option<String>,
954        metadata_entries: BTreeMap<String, String>,
955    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
956        let kind = kind.into();
957        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
958        if let Some(projection_name) = projection.as_deref() {
959            if !self.graph_projection_is_declared(projection_name) {
960                return Err(format!(
961                    "graph projection '{projection_name}' is not declared in physical metadata"
962                )
963                .into());
964            }
965            if !self.graph_projection_is_operational(projection_name) {
966                return Err(format!(
967                    "graph projection '{projection_name}' is declared but not operationally materialized"
968                )
969                .into());
970            }
971        }
972
973        self.update_physical_metadata(|metadata| {
974            let now = SystemTime::now()
975                .duration_since(UNIX_EPOCH)
976                .unwrap_or_default()
977                .as_millis();
978
979            let existing = metadata
980                .analytics_jobs
981                .iter_mut()
982                .find(|job| job.id == job_id)?;
983            existing.state = "completed".to_string();
984            existing.kind = kind.clone();
985            existing.projection = projection.clone();
986            existing.updated_at_unix_ms = now;
987            existing.last_run_sequence = Some(metadata.superblock.sequence);
988            existing.metadata = metadata_entries.clone();
989            let job = existing.clone();
990
991            metadata
992                .analytics_jobs
993                .sort_by(|left, right| left.id.cmp(&right.id));
994            Some(job)
995        })
996        .and_then(|job| {
997            job.ok_or_else(|| {
998                format!("analytics job '{job_id}' is not declared in physical metadata").into()
999            })
1000        })
1001    }
1002
1003    /// Mark a declared analytics job as running.
1004    pub fn queue_analytics_job(
1005        &self,
1006        kind: impl Into<String>,
1007        projection: Option<String>,
1008        metadata_entries: BTreeMap<String, String>,
1009    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1010        let kind = kind.into();
1011        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1012        if let Some(projection_name) = projection.as_deref() {
1013            if !self.graph_projection_is_declared(projection_name) {
1014                return Err(format!(
1015                    "graph projection '{projection_name}' is not declared in physical metadata"
1016                )
1017                .into());
1018            }
1019            if !self.graph_projection_is_operational(projection_name) {
1020                return Err(format!(
1021                    "graph projection '{projection_name}' is declared but not operationally materialized"
1022                )
1023                .into());
1024            }
1025        }
1026
1027        self.update_physical_metadata(|metadata| {
1028            let now = SystemTime::now()
1029                .duration_since(UNIX_EPOCH)
1030                .unwrap_or_default()
1031                .as_millis();
1032
1033            let existing = metadata
1034                .analytics_jobs
1035                .iter_mut()
1036                .find(|job| job.id == job_id)?;
1037            existing.state = "queued".to_string();
1038            existing.kind = kind.clone();
1039            existing.projection = projection.clone();
1040            existing.updated_at_unix_ms = now;
1041            existing.metadata = metadata_entries.clone();
1042            let job = existing.clone();
1043
1044            metadata
1045                .analytics_jobs
1046                .sort_by(|left, right| left.id.cmp(&right.id));
1047            Some(job)
1048        })
1049        .and_then(|job| {
1050            job.ok_or_else(|| {
1051                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1052            })
1053        })
1054    }
1055
1056    /// Mark a declared analytics job as running.
1057    pub fn start_analytics_job(
1058        &self,
1059        kind: impl Into<String>,
1060        projection: Option<String>,
1061        metadata_entries: BTreeMap<String, String>,
1062    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1063        let kind = kind.into();
1064        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1065        if let Some(projection_name) = projection.as_deref() {
1066            if !self.graph_projection_is_declared(projection_name) {
1067                return Err(format!(
1068                    "graph projection '{projection_name}' is not declared in physical metadata"
1069                )
1070                .into());
1071            }
1072            if !self.graph_projection_is_operational(projection_name) {
1073                return Err(format!(
1074                    "graph projection '{projection_name}' is declared but not operationally materialized"
1075                )
1076                .into());
1077            }
1078        }
1079
1080        self.update_physical_metadata(|metadata| {
1081            let now = SystemTime::now()
1082                .duration_since(UNIX_EPOCH)
1083                .unwrap_or_default()
1084                .as_millis();
1085
1086            let existing = metadata
1087                .analytics_jobs
1088                .iter_mut()
1089                .find(|job| job.id == job_id)?;
1090            existing.state = "running".to_string();
1091            existing.kind = kind.clone();
1092            existing.projection = projection.clone();
1093            existing.updated_at_unix_ms = now;
1094            existing.metadata = metadata_entries.clone();
1095            let job = existing.clone();
1096
1097            metadata
1098                .analytics_jobs
1099                .sort_by(|left, right| left.id.cmp(&right.id));
1100            Some(job)
1101        })
1102        .and_then(|job| {
1103            job.ok_or_else(|| {
1104                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1105            })
1106        })
1107    }
1108
1109    /// Mark a declared analytics job as failed.
1110    pub fn fail_analytics_job(
1111        &self,
1112        kind: impl Into<String>,
1113        projection: Option<String>,
1114        metadata_entries: BTreeMap<String, String>,
1115    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1116        let kind = kind.into();
1117        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1118
1119        self.update_physical_metadata(|metadata| {
1120            let now = SystemTime::now()
1121                .duration_since(UNIX_EPOCH)
1122                .unwrap_or_default()
1123                .as_millis();
1124
1125            let existing = metadata
1126                .analytics_jobs
1127                .iter_mut()
1128                .find(|job| job.id == job_id)?;
1129            existing.state = "failed".to_string();
1130            existing.kind = kind.clone();
1131            existing.projection = projection.clone();
1132            existing.updated_at_unix_ms = now;
1133            existing.metadata = metadata_entries.clone();
1134            let job = existing.clone();
1135
1136            metadata
1137                .analytics_jobs
1138                .sort_by(|left, right| left.id.cmp(&right.id));
1139            Some(job)
1140        })
1141        .and_then(|job| {
1142            job.ok_or_else(|| {
1143                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1144            })
1145        })
1146    }
1147
1148    /// Mark a declared analytics job as stale.
1149    pub fn mark_analytics_job_stale(
1150        &self,
1151        kind: impl Into<String>,
1152        projection: Option<String>,
1153        metadata_entries: BTreeMap<String, String>,
1154    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1155        let kind = kind.into();
1156        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1157
1158        self.update_physical_metadata(|metadata| {
1159            let now = SystemTime::now()
1160                .duration_since(UNIX_EPOCH)
1161                .unwrap_or_default()
1162                .as_millis();
1163
1164            let existing = metadata
1165                .analytics_jobs
1166                .iter_mut()
1167                .find(|job| job.id == job_id)?;
1168            existing.state = "stale".to_string();
1169            existing.kind = kind.clone();
1170            existing.projection = projection.clone();
1171            existing.updated_at_unix_ms = now;
1172            existing.metadata = metadata_entries.clone();
1173            let job = existing.clone();
1174
1175            metadata
1176                .analytics_jobs
1177                .sort_by(|left, right| left.id.cmp(&right.id));
1178            Some(job)
1179        })
1180        .and_then(|job| {
1181            job.ok_or_else(|| {
1182                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1183            })
1184        })
1185    }
1186
1187    /// Create a named export by copying the current database file and metadata sidecar.
1188    pub fn create_named_export(
1189        &self,
1190        name: impl Into<String>,
1191    ) -> Result<ExportDescriptor, Box<dyn std::error::Error>> {
1192        let name = name.into();
1193        if self.options.mode != StorageMode::Persistent {
1194            return Err("exports require persistent mode".into());
1195        }
1196        let Some(path) = self.path() else {
1197            return Err("database path is not available".into());
1198        };
1199
1200        self.flush()?;
1201
1202        let _publication = self.store.vault_publication_lock.lock();
1203        let mut metadata = match self.load_or_bootstrap_physical_metadata(true) {
1204            Ok(metadata) => metadata,
1205            Err(_)
1206                if self.options.storage_profile.deploy_profile
1207                    == crate::storage::DeployProfile::Embedded
1208                    && self.options.storage_profile.packaging
1209                        == crate::storage::StoragePackaging::SingleFile =>
1210            {
1211                crate::physical::PhysicalMetadataFile::from_state(
1212                    self.options.clone(),
1213                    self.catalog_snapshot(),
1214                    self.physical_collection_roots(),
1215                    self.physical_index_state(),
1216                    None,
1217                )
1218            }
1219            Err(err) => return Err(err),
1220        };
1221        let export_data_path = reddb_file::copy_physical_export_data_file(path, &name)?;
1222        let export_metadata_path = PhysicalMetadataFile::metadata_path_for(&export_data_path);
1223        let export_metadata_binary_path =
1224            PhysicalMetadataFile::metadata_binary_path_for(&export_data_path);
1225
1226        let descriptor = ExportDescriptor {
1227            name: name.clone(),
1228            created_at_unix_ms: SystemTime::now()
1229                .duration_since(UNIX_EPOCH)
1230                .unwrap_or_default()
1231                .as_millis(),
1232            snapshot_id: metadata
1233                .snapshots
1234                .last()
1235                .map(|snapshot| snapshot.snapshot_id),
1236            superblock_sequence: metadata.superblock.sequence,
1237            data_path: export_data_path.display().to_string(),
1238            metadata_path: export_metadata_path.display().to_string(),
1239            collection_count: metadata.catalog.total_collections,
1240            total_entities: metadata.catalog.total_entities,
1241        };
1242
1243        metadata
1244            .exports
1245            .retain(|export| export.name != descriptor.name);
1246        metadata.exports.push(descriptor.clone());
1247        self.prune_export_registry(&mut metadata.exports);
1248        metadata.save_for_data_path(path)?;
1249        metadata.save_to_binary_path(&export_metadata_binary_path)?;
1250        metadata.save_to_path(&export_metadata_path)?;
1251
1252        Ok(descriptor)
1253    }
1254
1255    /// Enable or disable a physical index entry in the persisted registry.
1256    pub fn set_index_enabled(
1257        &self,
1258        name: &str,
1259        enabled: bool,
1260    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1261        let Some(status) = self.index_status(name) else {
1262            return Err(format!("index '{name}' is not present in catalog status").into());
1263        };
1264        if !status.declared {
1265            return Err(format!("index '{name}' is not declared in physical metadata").into());
1266        }
1267        self.update_physical_metadata(|metadata| {
1268            let now = SystemTime::now()
1269                .duration_since(UNIX_EPOCH)
1270                .unwrap_or_default()
1271                .as_millis();
1272            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1273                index.enabled = enabled;
1274                if !enabled {
1275                    index.build_state = "disabled".to_string();
1276                } else if index.build_state == "disabled" {
1277                    index.build_state = if index.artifact_root_page.is_some() {
1278                        "ready".to_string()
1279                    } else {
1280                        "declared-unbuilt".to_string()
1281                    };
1282                }
1283                index.last_refresh_ms = Some(now);
1284                return Some(index.clone());
1285            }
1286            None
1287        })
1288    }
1289
1290    /// Mark a declared physical index as building in the persisted registry.
1291    pub fn mark_index_building(
1292        &self,
1293        name: &str,
1294    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1295        let Some(status) = self.index_status(name) else {
1296            return Err(format!("index '{name}' is not present in catalog status").into());
1297        };
1298        if !status.declared {
1299            return Err(format!("index '{name}' is not declared in physical metadata").into());
1300        }
1301        if status.lifecycle_state == "disabled" {
1302            return Err(format!("index '{name}' is disabled").into());
1303        }
1304        self.update_physical_metadata(|metadata| {
1305            let now = SystemTime::now()
1306                .duration_since(UNIX_EPOCH)
1307                .unwrap_or_default()
1308                .as_millis();
1309            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1310                index.build_state = "building".to_string();
1311                index.last_refresh_ms = Some(now);
1312                return Some(index.clone());
1313            }
1314            None
1315        })
1316    }
1317
1318    /// Mark a declared physical index as failed in the persisted registry.
1319    pub fn fail_index(
1320        &self,
1321        name: &str,
1322    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1323        let Some(status) = self.index_status(name) else {
1324            return Err(format!("index '{name}' is not present in catalog status").into());
1325        };
1326        if !status.declared {
1327            return Err(format!("index '{name}' is not declared in physical metadata").into());
1328        }
1329        self.update_physical_metadata(|metadata| {
1330            let now = SystemTime::now()
1331                .duration_since(UNIX_EPOCH)
1332                .unwrap_or_default()
1333                .as_millis();
1334            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1335                index.build_state = "failed".to_string();
1336                index.last_refresh_ms = Some(now);
1337                return Some(index.clone());
1338            }
1339            None
1340        })
1341    }
1342
1343    /// Mark a declared physical index as stale in the persisted registry.
1344    pub fn mark_index_stale(
1345        &self,
1346        name: &str,
1347    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1348        let Some(status) = self.index_status(name) else {
1349            return Err(format!("index '{name}' is not present in catalog status").into());
1350        };
1351        if !status.declared {
1352            return Err(format!("index '{name}' is not declared in physical metadata").into());
1353        }
1354        self.update_physical_metadata(|metadata| {
1355            let now = SystemTime::now()
1356                .duration_since(UNIX_EPOCH)
1357                .unwrap_or_default()
1358                .as_millis();
1359            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1360                index.build_state = if index.enabled {
1361                    "stale".to_string()
1362                } else {
1363                    "disabled".to_string()
1364                };
1365                index.last_refresh_ms = Some(now);
1366                return Some(index.clone());
1367            }
1368            None
1369        })
1370    }
1371
1372    /// Mark a declared physical index as ready in the persisted registry.
1373    pub fn mark_index_ready(
1374        &self,
1375        name: &str,
1376    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1377        self.warmup_index(name)
1378    }
1379
1380    /// Mark a physical index as warmed up/refreshed in the persisted registry.
1381    pub fn warmup_index(
1382        &self,
1383        name: &str,
1384    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1385        let Some(status) = self.index_status(name) else {
1386            return Err(format!("index '{name}' is not present in catalog status").into());
1387        };
1388        if !status.declared {
1389            return Err(format!("index '{name}' is not declared in physical metadata").into());
1390        }
1391        if status.lifecycle_state == "disabled" {
1392            return Err(format!("index '{name}' is disabled").into());
1393        }
1394        if !status.operational {
1395            return Err(
1396                format!("index '{name}' is declared but not operationally materialized").into(),
1397            );
1398        }
1399        let warmed_artifact = self
1400            .physical_indexes()
1401            .into_iter()
1402            .find(|index| index.name == name)
1403            .map(|mut index| {
1404                self.warmup_native_vector_artifact_for_index(&index)?;
1405                self.apply_runtime_native_artifact_to_index_state(&mut index)?;
1406                Ok::<_, String>(index)
1407            })
1408            .transpose()
1409            .map_err(|err| -> Box<dyn std::error::Error> { err.into() })?;
1410
1411        self.update_physical_metadata(|metadata| {
1412            let now = SystemTime::now()
1413                .duration_since(UNIX_EPOCH)
1414                .unwrap_or_default()
1415                .as_millis();
1416            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1417                if let Some(warmed) = warmed_artifact.as_ref() {
1418                    index.entries = warmed.entries;
1419                    index.estimated_memory_bytes = warmed.estimated_memory_bytes;
1420                    index.backend = warmed.backend.clone();
1421                    index.build_state = "ready".to_string();
1422                }
1423                index.last_refresh_ms = Some(now);
1424                return Some(index.clone());
1425            }
1426            None
1427        })
1428    }
1429
1430    /// Rebuild physical index metadata from the current catalog, optionally restricted to one collection.
1431    pub fn rebuild_index_registry(
1432        &self,
1433        collection: Option<&str>,
1434    ) -> Result<Vec<PhysicalIndexState>, Box<dyn std::error::Error>> {
1435        let fresh = self.reconcile_index_states_with_native_artifacts(self.physical_index_state());
1436        self.update_physical_metadata(|metadata| {
1437            let now = SystemTime::now()
1438                .duration_since(UNIX_EPOCH)
1439                .unwrap_or_default()
1440                .as_millis();
1441
1442            let mut affected = Vec::new();
1443            let declared = metadata.indexes.clone();
1444            for declared_index in declared {
1445                let matches_collection = collection.is_none_or(|collection_name| {
1446                    declared_index.collection.as_deref() == Some(collection_name)
1447                });
1448                if !matches_collection {
1449                    continue;
1450                }
1451
1452                let mut rebuilt = fresh
1453                    .iter()
1454                    .find(|index| index.name == declared_index.name)
1455                    .cloned()
1456                    .unwrap_or_else(|| {
1457                        let mut index = declared_index.clone();
1458                        index.build_state = "declared-unbuilt".to_string();
1459                        index
1460                    });
1461                rebuilt.enabled = declared_index.enabled;
1462                rebuilt.artifact_kind = rebuilt
1463                    .artifact_kind
1464                    .or_else(|| declared_index.artifact_kind.clone());
1465                rebuilt.artifact_root_page = rebuilt
1466                    .artifact_root_page
1467                    .or(declared_index.artifact_root_page);
1468                rebuilt.artifact_checksum = rebuilt
1469                    .artifact_checksum
1470                    .or(declared_index.artifact_checksum);
1471                rebuilt.build_state =
1472                    Self::finalize_rebuilt_index_build_state(&declared_index, &rebuilt);
1473                rebuilt.last_refresh_ms = Some(now);
1474
1475                if let Some(existing) = metadata
1476                    .indexes
1477                    .iter_mut()
1478                    .find(|index| index.name == rebuilt.name)
1479                {
1480                    *existing = rebuilt.clone();
1481                } else {
1482                    metadata.indexes.push(rebuilt.clone());
1483                }
1484
1485                affected.push(rebuilt);
1486            }
1487
1488            affected
1489        })
1490    }
1491
1492    fn finalize_rebuilt_index_build_state(
1493        declared: &PhysicalIndexState,
1494        rebuilt: &PhysicalIndexState,
1495    ) -> String {
1496        if !rebuilt.enabled {
1497            return "disabled".to_string();
1498        }
1499
1500        if declared.build_state == "failed" || rebuilt.build_state == "failed" {
1501            return "failed".to_string();
1502        }
1503
1504        let native_artifact_family = Self::native_artifact_kind_for_index(rebuilt.kind).is_some();
1505        if native_artifact_family {
1506            if rebuilt.artifact_root_page.is_some() && rebuilt.artifact_checksum.is_some() {
1507                return "ready".to_string();
1508            }
1509            if declared.artifact_root_page.is_some()
1510                || declared.artifact_checksum.is_some()
1511                || declared.artifact_kind.is_some()
1512            {
1513                return "stale".to_string();
1514            }
1515            return "declared-unbuilt".to_string();
1516        }
1517
1518        if rebuilt.entries > 0 {
1519            return "ready".to_string();
1520        }
1521
1522        if matches!(
1523            declared.build_state.as_str(),
1524            "stale" | "artifact-published" | "registry-loaded"
1525        ) {
1526            return "stale".to_string();
1527        }
1528
1529        "declared-unbuilt".to_string()
1530    }
1531}