uqa_storage/catalog/
graph_identifiers.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Physical allocation addresses and export projection for graph registry records. Graph owns label validation and graphid allocation rules.
8
9use std::collections::BTreeSet;
10
11use sha2::{Digest, Sha256};
12
13use crate::{GraphEntityKind, KeyValueBatch, StorageBackendResult};
14
15#[derive(Clone, Copy)]
16pub struct GraphIdentifierNamespace {
17    scope: [u8; 32],
18    generation: [u8; 16],
19}
20
21impl GraphIdentifierNamespace {
22    pub fn new(scope: Option<&str>, generation: [u8; 16]) -> Self {
23        let mut digest = Sha256::new();
24        digest.update([u8::from(scope.is_some())]);
25        digest.update(scope.unwrap_or_default().as_bytes());
26        Self {
27            scope: digest.finalize().into(),
28            generation,
29        }
30    }
31
32    pub fn key(&self, tag: u8, kind: GraphEntityKind, prefix: u32, graph: [u8; 16]) -> [u8; 78] {
33        let mut key = [0; 78];
34        key[..8].copy_from_slice(b"UQAGRPH1");
35        key[8..40].copy_from_slice(&self.scope);
36        key[40..56].copy_from_slice(&self.generation);
37        key[56] = tag;
38        key[57] = match kind {
39            GraphEntityKind::Vertex => b'v',
40            GraphEntityKind::Edge => b'e',
41        };
42        key[58..62].copy_from_slice(&prefix.to_be_bytes());
43        key[62..].copy_from_slice(&graph);
44        key
45    }
46
47    /// Legacy names identify only the original incarnation. New graph creation supplies a random identity; rename retains the original identity in its registry.
48    pub fn legacy_graph_identity(&self, graph: &str) -> [u8; 16] {
49        let mut digest = Sha256::new();
50        digest.update(self.key(b'g', GraphEntityKind::Vertex, 0, [0; 16]));
51        digest.update(graph.as_bytes());
52        let mut identity = [0; 16];
53        identity.copy_from_slice(&digest.finalize()[..16]);
54        if identity == [0; 16] {
55            identity[15] = 1;
56        }
57        identity
58    }
59
60    pub fn label_key(&self, graph: [u8; 16]) -> [u8; 78] {
61        self.key(b'l', GraphEntityKind::Vertex, 0, graph)
62    }
63
64    /// Logical graph entity observations share the physical entity scope and clear generation, independently of named graph membership or table row identities.
65    pub fn serializable_entity_object(&self) -> [u8; 16] {
66        self.serializable_object(b'r')
67    }
68
69    /// Clearing a namespace invalidates even an absent read from any earlier generation of that physical scope.
70    pub fn serializable_scope_object(&self) -> [u8; 16] {
71        Self {
72            generation: [0; 16],
73            ..*self
74        }
75        .serializable_object(b'R')
76    }
77
78    fn serializable_object(&self, tag: u8) -> [u8; 16] {
79        let digest = Sha256::digest(self.key(tag, GraphEntityKind::Vertex, 0, [0; 16]));
80        let mut identity = [0; 16];
81        identity.copy_from_slice(&digest[..16]);
82        if identity == [0; 16] {
83            identity[15] = 1;
84        }
85        identity
86    }
87
88    pub fn sequence_key(&self, graph: [u8; 16], label: u32) -> [u8; 78] {
89        self.key(b's', GraphEntityKind::Vertex, label, graph)
90    }
91
92    pub fn entity_key(&self, kind: GraphEntityKind, prefix: u32) -> [u8; 78] {
93        self.key(b'e', kind, prefix, [0; 16])
94    }
95
96    pub fn entity_seed_key(&self, kind: GraphEntityKind, prefix: u32) -> [u8; 78] {
97        self.key(b'i', kind, prefix, [0; 16])
98    }
99
100    pub fn hint_key(&self, kind: GraphEntityKind) -> [u8; 78] {
101        self.key(b'h', kind, 0, [0; 16])
102    }
103
104    /// Preserve a supplied physical identity before its evaluated record is published. The wire identity has a two-byte prefix and six-byte ordinal; observing one row does not certify that legacy rows were seeded.
105    pub fn observe_entity(
106        &self,
107        batch: &mut dyn KeyValueBatch,
108        kind: GraphEntityKind,
109        id: u64,
110    ) -> StorageBackendResult<()> {
111        let bytes = id.to_be_bytes();
112        let prefix = u32::from(u16::from_be_bytes([bytes[0], bytes[1]]));
113        let ordinal = id & (u64::MAX >> 16);
114        batch.observe_identifier(&self.entity_key(kind, prefix), ordinal)?;
115        batch.observe_identifier(&self.hint_key(kind), id)
116    }
117
118    /// Import durable counters encoded in a graph registry without assigning labels or evaluating graph behavior. Opaque legacy payloads retain their existing catalog contract.
119    pub fn observe_registry(
120        &self,
121        batch: &mut dyn KeyValueBatch,
122        graph: &str,
123        source: &str,
124    ) -> StorageBackendResult<()> {
125        let Ok(value) = serde_json::from_str::<serde_json::Value>(source) else {
126            return Ok(());
127        };
128        let Some(object) = value.as_object() else {
129            return Ok(());
130        };
131        let identity = object
132            .get("allocation_id")
133            .map(|value| serde_json::from_value::<[u8; 16]>(value.clone()))
134            .transpose()?
135            .filter(|id| *id != [0; 16])
136            .unwrap_or_else(|| self.legacy_graph_identity(graph));
137        if let Some(next) = object
138            .get("next_label_id")
139            .and_then(serde_json::Value::as_u64)
140        {
141            batch.observe_identifier(&self.label_key(identity), next.saturating_sub(1))?;
142        }
143        if let Some(sequences) = object
144            .get("sequences")
145            .and_then(serde_json::Value::as_object)
146        {
147            for (label, counter) in sequences {
148                let (Ok(label), Some(last)) = (label.parse::<u32>(), counter.as_u64()) else {
149                    continue;
150                };
151                batch.observe_identifier(&self.sequence_key(identity, label), last)?;
152                if let Ok(prefix) = u16::try_from(label) {
153                    if last != 0 && last <= (u64::MAX >> 16) {
154                        let mut bytes = last.to_be_bytes();
155                        bytes[..2].copy_from_slice(&prefix.to_be_bytes());
156                        let id = u64::from_be_bytes(bytes);
157                        let kind = registry_kind(&value, label);
158                        for candidate in [GraphEntityKind::Vertex, GraphEntityKind::Edge] {
159                            if kind.is_none_or(|kind| kind == candidate) {
160                                self.observe_entity(batch, candidate, id)?;
161                            }
162                        }
163                    }
164                }
165            }
166        }
167        Ok(())
168    }
169}
170
171fn registry_kind(value: &serde_json::Value, label: u32) -> Option<GraphEntityKind> {
172    match label {
173        1 => Some(GraphEntityKind::Vertex),
174        2 => Some(GraphEntityKind::Edge),
175        _ => value
176            .get("labels")?
177            .as_object()?
178            .iter()
179            .find_map(|(name, id)| {
180                if id.as_u64() != Some(u64::from(label)) {
181                    return None;
182                }
183                match value
184                    .get("kinds")
185                    .and_then(|kinds| kinds.get(name))
186                    .and_then(serde_json::Value::as_str)
187                {
188                    Some("v") => Some(GraphEntityKind::Vertex),
189                    Some("e") => Some(GraphEntityKind::Edge),
190                    _ => None,
191                }
192            }),
193    }
194}
195
196/// Copy current autonomous floors into an explicit graph export. Unknown fields and unchanged legacy payloads retain their original representation; this does not publish a catalog write or alter its pinned definition.
197pub fn export_registry(
198    source: &str,
199    graph: &str,
200    namespace: GraphIdentifierNamespace,
201    mut watermark: impl FnMut(&[u8]) -> StorageBackendResult<Option<u64>>,
202) -> StorageBackendResult<String> {
203    let mut value: serde_json::Value = if source.is_empty() {
204        serde_json::json!({})
205    } else {
206        match serde_json::from_str(source) {
207            Ok(value) => value,
208            Err(_) => return Ok(source.to_owned()),
209        }
210    };
211    let Some(object) = value.as_object_mut() else {
212        return Ok(source.to_owned());
213    };
214    let identity = object
215        .get("allocation_id")
216        .map(|value| serde_json::from_value::<[u8; 16]>(value.clone()))
217        .transpose()?
218        .filter(|id| *id != [0; 16])
219        .unwrap_or_else(|| namespace.legacy_graph_identity(graph));
220    let mut changed = false;
221    if let Some(last) = watermark(&namespace.label_key(identity))? {
222        let next = last.checked_add(1).ok_or_else(|| {
223            crate::StorageBackendError::Other("graph label watermark overflow".into())
224        })?;
225        if object
226            .get("next_label_id")
227            .and_then(serde_json::Value::as_u64)
228            .is_none_or(|old| old < next)
229        {
230            object.insert("next_label_id".into(), next.into());
231            changed = true;
232        }
233    }
234    let mut labels = BTreeSet::from([1_u32, 2]);
235    if let Some(registered) = object.get("labels").and_then(serde_json::Value::as_object) {
236        for value in registered.values() {
237            if let Some(label) = value.as_u64().and_then(|id| u32::try_from(id).ok()) {
238                labels.insert(label);
239            }
240        }
241    }
242    if let Some(sequences) = object
243        .get("sequences")
244        .and_then(serde_json::Value::as_object)
245    {
246        for label in sequences.keys() {
247            if let Ok(label) = label.parse() {
248                labels.insert(label);
249            }
250        }
251    }
252    for label in labels {
253        let Some(last) = watermark(&namespace.sequence_key(identity, label))? else {
254            continue;
255        };
256        let sequences = object
257            .entry("sequences")
258            .or_insert_with(|| serde_json::json!({}));
259        let sequences = sequences.as_object_mut().ok_or_else(|| {
260            crate::StorageBackendError::Other("invalid graph registry sequences".into())
261        })?;
262        let key = label.to_string();
263        if sequences
264            .get(&key)
265            .and_then(serde_json::Value::as_u64)
266            .is_none_or(|old| old < last)
267        {
268            sequences.insert(key, last.into());
269            changed = true;
270        }
271    }
272    if changed {
273        object.insert("allocation_id".into(), serde_json::to_value(identity)?);
274        Ok(serde_json::to_string(&value)?)
275    } else {
276        Ok(source.to_owned())
277    }
278}
uqa_storage/catalog/graph_identifiers.rs

uqa_storage/catalog/
graph_identifiers.rs
