graphforge_knowledge/
source.rs

1//! Immutable research Source identities and Arrow encoding (#1349).
2
3use std::collections::HashSet;
4use std::sync::{Arc, LazyLock};
5
6use arrow::array::{FixedSizeBinaryBuilder, StringArray, TimestampMicrosecondArray, UInt32Array};
7use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
8use arrow::record_batch::RecordBatch;
9use graphforge_core::canonical::{
10    CANONICAL_CONTRACT_VERSION, CanonicalDomain, CanonicalWriter, fingerprint,
11};
12use uuid::Uuid;
13
14use crate::{
15    KNOWLEDGE_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
16    check_limit, fixed_column, invalid, optional_text, require_schema, require_uuid, require_v7,
17    required_i64, required_text, required_u32, string_column, timestamp_column, u32_column,
18    uuid_at, uuid_field,
19};
20
21/// Immutable research Source record contract.
22pub const SOURCE_CONTRACT_VERSION: u32 = 1;
23/// Closed research-source-kind registry version.
24pub const SOURCE_KIND_REGISTRY_VERSION: u32 = 1;
25/// Maximum UTF-8 bytes for one bounded Source label.
26pub const MAX_SOURCE_LABEL_BYTES: usize = 4_096;
27/// Maximum UTF-8 bytes for one bounded external identity URI.
28pub const MAX_SOURCE_IDENTITY_URI_BYTES: usize = 4_096;
29
30/// Authoritative `knowledge/sources.parquet` schema.
31pub static SOURCE_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
32    Arc::new(Schema::new(vec![
33        uuid_field("source_uuid", false),
34        Field::new("label", DataType::Utf8, false),
35        Field::new("source_kind", DataType::Utf8, false),
36        Field::new("identity_uri", DataType::Utf8, true),
37        uuid_field("provenance_uuid", false),
38        Field::new(
39            "recorded_at",
40            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
41            false,
42        ),
43        Field::new("contract_version", DataType::UInt32, false),
44    ]))
45});
46
47static SOURCE_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
48    fingerprint(
49        CanonicalDomain::Schema,
50        CANONICAL_CONTRACT_VERSION,
51        b"source/1|source_uuid:fixed[16]:required|label:utf8:required|source_kind:utf8:required|identity_uri:utf8:nullable|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
52    )
53    .expect("registered source schema is within canonical bounds")
54});
55
56/// Closed kind for one acquired research Source.
57#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)]
58#[serde(rename_all = "snake_case")]
59pub enum SourceKind {
60    /// Handwritten or typed manuscript.
61    Manuscript,
62    /// Published edition identity.
63    Edition,
64    /// Portable document representation.
65    Pdf,
66    /// Electronic publication package.
67    Epub,
68    /// Web resource or canonical URL identity.
69    Web,
70    /// Photograph or still image capture.
71    Photograph,
72    /// Audio or audiovisual recording.
73    Recording,
74    /// Structured database export snapshot.
75    DatabaseExport,
76    /// Other closed kind not listed above.
77    Other,
78}
79
80impl SourceKind {
81    /// Canonical persisted spelling.
82    #[must_use]
83    pub const fn as_str(self) -> &'static str {
84        match self {
85            Self::Manuscript => "manuscript",
86            Self::Edition => "edition",
87            Self::Pdf => "pdf",
88            Self::Epub => "epub",
89            Self::Web => "web",
90            Self::Photograph => "photograph",
91            Self::Recording => "recording",
92            Self::DatabaseExport => "database_export",
93            Self::Other => "other",
94        }
95    }
96
97    pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
98        match value {
99            "manuscript" => Ok(Self::Manuscript),
100            "edition" => Ok(Self::Edition),
101            "pdf" => Ok(Self::Pdf),
102            "epub" => Ok(Self::Epub),
103            "web" => Ok(Self::Web),
104            "photograph" => Ok(Self::Photograph),
105            "recording" => Ok(Self::Recording),
106            "database_export" => Ok(Self::DatabaseExport),
107            "other" => Ok(Self::Other),
108            _ => Err(invalid("source_kind", "unknown closed value")),
109        }
110    }
111}
112
113/// One immutable research Source identity.
114#[derive(Clone, Debug, Eq, PartialEq)]
115pub struct Source {
116    /// Caller-supplied UUIDv7 identity and idempotency key.
117    pub source_uuid: Uuid,
118    /// Bounded human-readable label; never raw source text.
119    pub label: String,
120    /// Closed source kind.
121    pub source_kind: SourceKind,
122    /// Stable external identity such as DOI or canonical URL; never fetched.
123    pub identity_uri: Option<String>,
124    /// Provenance event that published the Source.
125    pub provenance_uuid: Uuid,
126    /// Durable transaction time.
127    pub recorded_at_micros: i64,
128    /// Frozen record contract.
129    pub contract_version: u32,
130}
131
132impl Source {
133    /// Construct one validated immutable Source.
134    pub fn new(
135        source_uuid: Uuid,
136        label: String,
137        source_kind: SourceKind,
138        identity_uri: Option<String>,
139        provenance_uuid: Uuid,
140        recorded_at_micros: i64,
141    ) -> Result<Self, KnowledgeError> {
142        let row = Self {
143            source_uuid,
144            label,
145            source_kind,
146            identity_uri,
147            provenance_uuid,
148            recorded_at_micros,
149            contract_version: SOURCE_CONTRACT_VERSION,
150        };
151        validate_source(&row)?;
152        Ok(row)
153    }
154}
155
156/// Validated immutable Source table.
157#[derive(Clone, Debug, Default, Eq, PartialEq)]
158pub struct SourceLedger {
159    /// Sources ordered by `(recorded_at, source_uuid)`.
160    pub sources: Vec<Source>,
161}
162
163impl SourceLedger {
164    /// Validate, sort, and construct a Source table.
165    pub fn new(mut sources: Vec<Source>) -> Result<Self, KnowledgeError> {
166        sources.sort_by_key(|row| (row.recorded_at_micros, row.source_uuid));
167        validate_sources(&sources)?;
168        Ok(Self { sources })
169    }
170
171    /// Merge staged rows into an existing ledger.
172    pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
173        let mut merged = self.sources.clone();
174        for row in &staged.sources {
175            if let Some(existing) = merged
176                .iter()
177                .find(|candidate| candidate.source_uuid == row.source_uuid)
178            {
179                if existing != row {
180                    return Err(KnowledgeError::Conflict("source_uuid"));
181                }
182                continue;
183            }
184            merged.push(row.clone());
185        }
186        Self::new(merged)
187    }
188
189    /// Canonical semantic fingerprint over immutable Source content, excluding clock fields.
190    ///
191    /// Excludes `provenance_uuid` and `recorded_at_micros` so that a retry with the same
192    /// semantic content but a different clock value compares equal to the committed row.
193    pub fn source_fingerprint(&self, source_uuid: Uuid) -> Result<[u8; 32], KnowledgeError> {
194        let row = self
195            .sources
196            .iter()
197            .find(|row| row.source_uuid == source_uuid)
198            .ok_or(KnowledgeError::Dangling("source_uuid"))?;
199        let mut writer = CanonicalWriter::new();
200        writer.raw(b"GFSO")?;
201        writer.u32(SOURCE_CONTRACT_VERSION)?;
202        writer.raw(row.source_uuid.as_bytes())?;
203        writer.text(&row.label)?;
204        writer.text(row.source_kind.as_str())?;
205        match &row.identity_uri {
206            None => writer.u8(0)?,
207            Some(uri) => {
208                writer.u8(1)?;
209                writer.text(uri)?;
210            }
211        }
212        Ok(fingerprint(
213            CanonicalDomain::ResearchSource,
214            CANONICAL_CONTRACT_VERSION,
215            &writer.finish(),
216        )?)
217    }
218
219    /// Encode the authoritative Arrow batch.
220    pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
221        source_batch(&self.sources)
222    }
223
224    /// Decode one or more Arrow batches.
225    pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
226        let mut sources = Vec::new();
227        for batch in batches {
228            require_schema(batch, &SOURCE_SCHEMA, "sources.schema")?;
229            let ids = fixed_column(batch, "source_uuid")?;
230            let labels = string_column(batch, "label")?;
231            let kinds = string_column(batch, "source_kind")?;
232            let uris = string_column(batch, "identity_uri")?;
233            let provenance = fixed_column(batch, "provenance_uuid")?;
234            let recorded = timestamp_column(batch, "recorded_at")?;
235            let versions = u32_column(batch, "contract_version")?;
236            for row in 0..batch.num_rows() {
237                sources.push(Source {
238                    source_uuid: uuid_at(ids, row, "source_uuid")?,
239                    label: required_text(labels, row, "label")?.to_string(),
240                    source_kind: SourceKind::parse(required_text(kinds, row, "source_kind")?)?,
241                    identity_uri: optional_text(uris, row),
242                    provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
243                    recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
244                    contract_version: required_u32(versions, row, "contract_version")?,
245                });
246            }
247        }
248        Self::new(sources)
249    }
250}
251
252pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
253    SchemaRegistryEntry {
254        capability_id: "knowledge",
255        capability_version: KNOWLEDGE_CAPABILITY_VERSION,
256        record_family: "sources",
257        record_version: SOURCE_CONTRACT_VERSION,
258        schema: Arc::clone(&SOURCE_SCHEMA),
259        schema_fingerprint: *SOURCE_SCHEMA_FINGERPRINT,
260        enum_registry_versions: &[("source_kind", SOURCE_KIND_REGISTRY_VERSION)],
261        sort_key: &["recorded_at", "source_uuid"],
262        diff_identity_fields: &["source_uuid"],
263        diff_record_uuid_field: Some("source_uuid"),
264        fingerprint_domain: CanonicalDomain::ResearchSource,
265        owner: "graphforge-knowledge",
266        implementation_issue: 1349,
267        max_rows: MAX_KNOWLEDGE_ROWS,
268    }
269}
270
271fn validate_source(row: &Source) -> Result<(), KnowledgeError> {
272    if row.contract_version != SOURCE_CONTRACT_VERSION {
273        return Err(invalid("source.contract_version", "unsupported version"));
274    }
275    require_v7(row.source_uuid, "source_uuid")?;
276    require_uuid(row.provenance_uuid, "provenance_uuid")?;
277    validate_bounded_text("source.label", &row.label, MAX_SOURCE_LABEL_BYTES)?;
278    if let Some(uri) = &row.identity_uri {
279        validate_bounded_text("source.identity_uri", uri, MAX_SOURCE_IDENTITY_URI_BYTES)?;
280    }
281    Ok(())
282}
283
284fn validate_sources(sources: &[Source]) -> Result<(), KnowledgeError> {
285    check_limit("sources", sources.len())?;
286    let mut ids = HashSet::with_capacity(sources.len());
287    for row in sources {
288        validate_source(row)?;
289        if !ids.insert(row.source_uuid) {
290            return Err(KnowledgeError::Duplicate("source_uuid"));
291        }
292    }
293    Ok(())
294}
295
296fn validate_bounded_text(
297    field: &'static str,
298    value: &str,
299    limit: usize,
300) -> Result<(), KnowledgeError> {
301    if value.is_empty() {
302        return Err(invalid(field, "value must not be empty"));
303    }
304    if value.len() > limit {
305        return Err(KnowledgeError::Limit {
306            participant: field,
307            observed: value.len(),
308            limit,
309        });
310    }
311    Ok(())
312}
313
314fn source_batch(rows: &[Source]) -> Result<RecordBatch, KnowledgeError> {
315    let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
316    let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
317    for row in rows {
318        ids.append_value(row.source_uuid.as_bytes())?;
319        provenance.append_value(row.provenance_uuid.as_bytes())?;
320    }
321    RecordBatch::try_new(
322        Arc::clone(&SOURCE_SCHEMA),
323        vec![
324            Arc::new(ids.finish()),
325            Arc::new(StringArray::from_iter_values(
326                rows.iter().map(|row| row.label.as_str()),
327            )),
328            Arc::new(StringArray::from_iter_values(
329                rows.iter().map(|row| row.source_kind.as_str()),
330            )),
331            Arc::new(StringArray::from(
332                rows.iter()
333                    .map(|row| row.identity_uri.as_deref())
334                    .collect::<Vec<_>>(),
335            )),
336            Arc::new(provenance.finish()),
337            Arc::new(
338                TimestampMicrosecondArray::from_iter_values(
339                    rows.iter().map(|row| row.recorded_at_micros),
340                )
341                .with_timezone("UTC"),
342            ),
343            Arc::new(UInt32Array::from_iter_values(
344                rows.iter().map(|row| row.contract_version),
345            )),
346        ],
347    )
348    .map_err(Into::into)
349}
350
351#[cfg(test)]
352mod tests {
353    use super::*;
354    use crate::tests::uuid7;
355
356    #[test]
357    fn source_round_trip_and_merge_idempotency() {
358        let first = Source::new(
359            uuid7(1),
360            "Arabian Nights manuscript".into(),
361            SourceKind::Manuscript,
362            Some("https://example.org/ark:/12345".into()),
363            uuid7(2),
364            100,
365        )
366        .unwrap();
367        let ledger = SourceLedger::new(vec![first.clone()]).unwrap();
368        let batch = ledger.batch().unwrap();
369        let reopened = SourceLedger::from_batches(&[batch]).unwrap();
370        assert_eq!(reopened, ledger);
371        let duplicate = SourceLedger::new(vec![first.clone()]).unwrap();
372        assert_eq!(ledger.merge(&duplicate).unwrap(), ledger);
373        let conflicting = Source::new(
374            first.source_uuid,
375            "Different label".into(),
376            SourceKind::Pdf,
377            None,
378            uuid7(3),
379            200,
380        )
381        .unwrap();
382        assert!(matches!(
383            ledger.merge(&SourceLedger::new(vec![conflicting]).unwrap()),
384            Err(KnowledgeError::Conflict("source_uuid"))
385        ));
386    }
387
388    #[test]
389    fn source_kind_registry_round_trip_and_reject_unknown_tokens() {
390        for kind in [
391            SourceKind::Manuscript,
392            SourceKind::Edition,
393            SourceKind::Pdf,
394            SourceKind::Epub,
395            SourceKind::Web,
396            SourceKind::Photograph,
397            SourceKind::Recording,
398            SourceKind::DatabaseExport,
399            SourceKind::Other,
400        ] {
401            assert_eq!(SourceKind::parse(kind.as_str()).unwrap(), kind);
402        }
403        assert!(SourceKind::parse("scan").is_err());
404    }
405}
graphforge_knowledge/source.rs

graphforge_knowledge/
source.rs
