knowledge_runtime/projection/
lifecycle.rs

1use crate::ids::{ProjectionId, ProjectionKind, ScopeKey};
2use chrono::{DateTime, Utc};
3use serde::{Deserialize, Serialize};
4use std::collections::BTreeMap;
5
6/// Health status of a projection.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(rename_all = "snake_case")]
9pub enum ProjectionHealth {
10    /// Projection is current and healthy.
11    Healthy,
12    /// Projection exists but may be stale.
13    Stale,
14    /// Projection is missing or failed to build.
15    Missing,
16    /// Reserved compatibility state for external rebuild orchestration.
17    ///
18    /// The in-memory tracker does not currently emit this variant on its own.
19    Rebuilding,
20    /// Reserved compatibility state for import-aware callers.
21    ///
22    /// The runtime currently surfaces import freshness as `ProjectionImportStale`
23    /// warnings in query traces rather than transitioning tracker health here.
24    ImportLagging,
25    /// Reserved compatibility state for import-aware callers.
26    ///
27    /// The tracker does not currently transition into this state.
28    ImportFailed,
29}
30
31/// Why a projection became stale.
32#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum StaleCause {
35    /// Staleness threshold exceeded (time-based).
36    TimeThreshold,
37    /// Explicitly invalidated via API.
38    ExplicitInvalidation { reason: String },
39    /// Source data changed upstream.
40    SourceChanged,
41    /// Schema or resolver version changed.
42    VersionMismatch,
43    /// Upstream import pipeline is lagging behind source.
44    ImportLag {
45        /// ISO 8601 timestamp of the last successful import, if any.
46        last_import_at: Option<String>,
47    },
48    /// The last import attempt failed.
49    ImportFailure {
50        /// Error from the failed import.
51        error: String,
52    },
53}
54
55/// Version metadata for a projection.
56///
57/// Tracks schema and resolver versions so that version changes can trigger
58/// invalidation. Fields are opaque strings — the runtime does not interpret
59/// them, just compares for equality.
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
61pub struct ProjectionVersion {
62    /// Schema version of the projection data format.
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub schema_version: Option<String>,
65    /// Version of the resolver/adapter that built this projection.
66    #[serde(skip_serializing_if = "Option::is_none")]
67    pub resolver_version: Option<String>,
68}
69
70/// Metadata about a projection's lifecycle.
71#[derive(Debug, Clone, Serialize, Deserialize)]
72pub struct ProjectionMeta {
73    /// Projection identity.
74    pub id: ProjectionId,
75    /// Current health.
76    pub health: ProjectionHealth,
77    /// Why the projection is stale (if applicable).
78    #[serde(skip_serializing_if = "Option::is_none")]
79    pub stale_cause: Option<StaleCause>,
80    /// When the projection was last built.
81    #[serde(skip_serializing_if = "Option::is_none")]
82    pub built_at: Option<DateTime<Utc>>,
83    /// When the projection was last invalidated.
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub invalidated_at: Option<DateTime<Utc>>,
86    /// How long the last build took, in milliseconds.
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub build_duration_ms: Option<u64>,
89    /// Number of source items that fed the projection.
90    #[serde(skip_serializing_if = "Option::is_none")]
91    pub source_count: Option<usize>,
92    /// Last error if the projection is unhealthy.
93    #[serde(skip_serializing_if = "Option::is_none")]
94    pub last_error: Option<String>,
95    /// Version metadata.
96    #[serde(skip_serializing_if = "Option::is_none")]
97    pub version: Option<ProjectionVersion>,
98}
99
100/// An invalidation signal that marks projections as stale.
101#[derive(Debug, Clone, Serialize, Deserialize)]
102pub struct InvalidationEvent {
103    /// Which projections are affected.
104    pub projection_ids: Vec<ProjectionId>,
105    /// Why they are being invalidated.
106    pub cause: StaleCause,
107    /// When the invalidation was issued.
108    pub at: DateTime<Utc>,
109}
110
111/// Result of a projection lifecycle action.
112#[derive(Debug, Clone, Serialize, Deserialize)]
113pub struct ProjectionActionResult {
114    /// How many projections were affected.
115    pub affected_count: usize,
116    /// IDs of affected projections.
117    pub affected_ids: Vec<ProjectionId>,
118}
119
120/// Projection lifecycle tracker.
121///
122/// Tracks the health and staleness metadata of all projections.
123/// Projections are identified by `ProjectionId` which includes scope.
124///
125/// ## Authority: Observability only, NOT orchestration
126///
127/// This tracker is a **passive bookkeeping / observability component**.
128/// It records state transitions (build, invalidation, failure) so that
129/// callers can inspect projection health and decide whether to trigger
130/// rebuilds. Current methods emit `Healthy`, `Stale`, and `Missing`;
131/// other `ProjectionHealth` variants remain reserved for compatibility
132/// and external orchestration surfaces. The tracker does NOT:
133///
134/// - Schedule or execute projection rebuilds
135/// - Retry failed builds automatically
136/// - Communicate with upstream import pipelines
137/// - Own any source of truth
138///
139/// Callers are responsible for executing rebuilds and reporting
140/// outcomes via `record_build()` / `record_failure()`.
141#[derive(Debug)]
142pub struct ProjectionTracker {
143    /// Projection metadata by ID.
144    projections: BTreeMap<ProjectionId, ProjectionMeta>,
145    /// Staleness threshold in seconds.
146    staleness_threshold_secs: u64,
147}
148
149impl ProjectionTracker {
150    pub fn new(staleness_threshold_secs: u64) -> Self {
151        Self {
152            projections: BTreeMap::new(),
153            staleness_threshold_secs,
154        }
155    }
156
157    /// Record that a projection was successfully built.
158    ///
159    /// **This is a bookkeeping method, not a rebuild trigger.** The tracker
160    /// records that a build occurred but does not execute, schedule, or
161    /// orchestrate any rebuild work. External callers must drive the actual
162    /// rebuild and call this method to report the outcome.
163    pub fn record_build(
164        &mut self,
165        id: ProjectionId,
166        source_count: usize,
167        build_duration_ms: u64,
168        version: Option<ProjectionVersion>,
169    ) {
170        self.projections.insert(
171            id.clone(),
172            ProjectionMeta {
173                id,
174                health: ProjectionHealth::Healthy,
175                stale_cause: None,
176                built_at: Some(Utc::now()),
177                invalidated_at: None,
178                build_duration_ms: Some(build_duration_ms),
179                source_count: Some(source_count),
180                last_error: None,
181                version,
182            },
183        );
184    }
185
186    /// Record that a projection build failed.
187    ///
188    /// **This is a bookkeeping method, not a retry trigger.** The tracker
189    /// records the failure but does not retry, schedule, or orchestrate any
190    /// rebuild attempt. External callers must decide whether and when to
191    /// retry, and call `record_build()` or `record_failure()` to report
192    /// the next outcome.
193    pub fn record_failure(&mut self, id: ProjectionId, error: String) {
194        let existing = self.projections.get(&id);
195        self.projections.insert(
196            id.clone(),
197            ProjectionMeta {
198                id,
199                health: ProjectionHealth::Missing,
200                stale_cause: None,
201                built_at: existing.and_then(|p| p.built_at),
202                invalidated_at: None,
203                build_duration_ms: None,
204                source_count: existing.and_then(|p| p.source_count),
205                last_error: Some(error),
206                version: existing.and_then(|p| p.version.clone()),
207            },
208        );
209    }
210
211    /// Process an invalidation event, marking affected projections as stale.
212    pub fn invalidate(&mut self, event: &InvalidationEvent) -> ProjectionActionResult {
213        let mut affected = Vec::new();
214        let now = Utc::now();
215
216        for pid in &event.projection_ids {
217            if let Some(meta) = self.projections.get_mut(pid) {
218                meta.health = ProjectionHealth::Stale;
219                meta.stale_cause = Some(event.cause.clone());
220                meta.invalidated_at = Some(now);
221                affected.push(pid.clone());
222            }
223        }
224
225        ProjectionActionResult {
226            affected_count: affected.len(),
227            affected_ids: affected,
228        }
229    }
230
231    /// Invalidate all projections of a given kind within a scope.
232    pub fn invalidate_by_kind_and_scope(
233        &mut self,
234        kind: &ProjectionKind,
235        scope: &ScopeKey,
236        cause: StaleCause,
237    ) -> ProjectionActionResult {
238        let now = Utc::now();
239        let mut affected = Vec::new();
240
241        for (pid, meta) in self.projections.iter_mut() {
242            if &pid.kind == kind && &pid.scope == scope {
243                meta.health = ProjectionHealth::Stale;
244                meta.stale_cause = Some(cause.clone());
245                meta.invalidated_at = Some(now);
246                affected.push(pid.clone());
247            }
248        }
249
250        ProjectionActionResult {
251            affected_count: affected.len(),
252            affected_ids: affected,
253        }
254    }
255
256    /// Invalidate all projections within a scope (all kinds).
257    pub fn invalidate_scope(
258        &mut self,
259        scope: &ScopeKey,
260        cause: StaleCause,
261    ) -> ProjectionActionResult {
262        let now = Utc::now();
263        let mut affected = Vec::new();
264
265        for (pid, meta) in self.projections.iter_mut() {
266            if &pid.scope == scope {
267                meta.health = ProjectionHealth::Stale;
268                meta.stale_cause = Some(cause.clone());
269                meta.invalidated_at = Some(now);
270                affected.push(pid.clone());
271            }
272        }
273
274        ProjectionActionResult {
275            affected_count: affected.len(),
276            affected_ids: affected,
277        }
278    }
279
280    /// Get the health of a specific projection (with time-based staleness check).
281    pub fn health(&self, id: &ProjectionId) -> ProjectionHealth {
282        self.projections
283            .get(id)
284            .map(|m| {
285                if m.health == ProjectionHealth::Healthy {
286                    if let Some(built_at) = m.built_at {
287                        let age = Utc::now()
288                            .signed_duration_since(built_at)
289                            .num_seconds()
290                            .unsigned_abs();
291                        if age > self.staleness_threshold_secs {
292                            return ProjectionHealth::Stale;
293                        }
294                    }
295                }
296                m.health.clone()
297            })
298            .unwrap_or(ProjectionHealth::Missing)
299    }
300
301    /// Get metadata for a specific projection.
302    pub fn get(&self, id: &ProjectionId) -> Option<&ProjectionMeta> {
303        self.projections.get(id)
304    }
305
306    /// Iterate over all tracked projection IDs.
307    ///
308    /// Useful for external rebuild drivers that need to scan for stale projections.
309    pub fn all_ids(&self) -> impl Iterator<Item = &ProjectionId> {
310        self.projections.keys()
311    }
312
313    /// Query status of all projections matching a kind and/or scope filter.
314    pub fn query_status(
315        &self,
316        kind: Option<&ProjectionKind>,
317        scope: Option<&ScopeKey>,
318    ) -> Vec<&ProjectionMeta> {
319        self.projections
320            .values()
321            .filter(|m| {
322                kind.map_or(true, |k| &m.id.kind == k) && scope.map_or(true, |s| &m.id.scope == s)
323            })
324            .collect()
325    }
326
327    /// List all tracked projections.
328    pub fn list(&self) -> Vec<&ProjectionMeta> {
329        self.projections.values().collect()
330    }
331
332    /// Remove a projection from tracking (forces recomputation on next access).
333    pub fn remove(&mut self, id: &ProjectionId) -> Option<ProjectionMeta> {
334        self.projections.remove(id)
335    }
336
337    /// Clear all projections within a scope.
338    pub fn clear_scope(&mut self, scope: &ScopeKey) -> ProjectionActionResult {
339        let ids_to_remove: Vec<ProjectionId> = self
340            .projections
341            .keys()
342            .filter(|pid| &pid.scope == scope)
343            .cloned()
344            .collect();
345
346        let affected_count = ids_to_remove.len();
347        for id in &ids_to_remove {
348            self.projections.remove(id);
349        }
350
351        ProjectionActionResult {
352            affected_count,
353            affected_ids: ids_to_remove,
354        }
355    }
356}
357
358#[cfg(test)]
359mod tests {
360    use super::*;
361
362    fn test_scope() -> ScopeKey {
363        ScopeKey::namespace_only("test")
364    }
365
366    fn test_pid(key: &str) -> ProjectionId {
367        ProjectionId::new(ProjectionKind::Entity, key, test_scope())
368    }
369
370    #[test]
371    fn missing_by_default() {
372        let tracker = ProjectionTracker::new(3600);
373        assert_eq!(tracker.health(&test_pid("x")), ProjectionHealth::Missing);
374    }
375
376    #[test]
377    fn healthy_after_build() {
378        let mut tracker = ProjectionTracker::new(3600);
379        let id = test_pid("test");
380        tracker.record_build(id.clone(), 10, 50, None);
381        assert_eq!(tracker.health(&id), ProjectionHealth::Healthy);
382    }
383
384    #[test]
385    fn stale_after_invalidation() {
386        let mut tracker = ProjectionTracker::new(3600);
387        let id = test_pid("test");
388        tracker.record_build(id.clone(), 10, 50, None);
389
390        let event = InvalidationEvent {
391            projection_ids: vec![id.clone()],
392            cause: StaleCause::SourceChanged,
393            at: Utc::now(),
394        };
395        let result = tracker.invalidate(&event);
396        assert_eq!(result.affected_count, 1);
397        assert_eq!(tracker.health(&id), ProjectionHealth::Stale);
398
399        // Verify stale cause is recorded
400        let meta = tracker.get(&id).unwrap();
401        assert_eq!(meta.stale_cause, Some(StaleCause::SourceChanged));
402        assert!(meta.invalidated_at.is_some());
403    }
404
405    #[test]
406    fn invalidate_by_kind_and_scope() {
407        let mut tracker = ProjectionTracker::new(3600);
408        let scope = test_scope();
409        let other_scope = ScopeKey::namespace_only("other");
410
411        let id1 = ProjectionId::new(ProjectionKind::Entity, "a", scope.clone());
412        let id2 = ProjectionId::new(ProjectionKind::Entity, "b", scope.clone());
413        let id3 = ProjectionId::new(ProjectionKind::Entity, "c", other_scope.clone());
414        let id4 = ProjectionId::new(ProjectionKind::Temporal, "d", scope.clone());
415
416        tracker.record_build(id1.clone(), 5, 10, None);
417        tracker.record_build(id2.clone(), 5, 10, None);
418        tracker.record_build(id3.clone(), 5, 10, None);
419        tracker.record_build(id4.clone(), 5, 10, None);
420
421        let result = tracker.invalidate_by_kind_and_scope(
422            &ProjectionKind::Entity,
423            &scope,
424            StaleCause::ExplicitInvalidation {
425                reason: "test".into(),
426            },
427        );
428
429        assert_eq!(result.affected_count, 2);
430        assert_eq!(tracker.health(&id1), ProjectionHealth::Stale);
431        assert_eq!(tracker.health(&id2), ProjectionHealth::Stale);
432        assert_eq!(tracker.health(&id3), ProjectionHealth::Healthy); // different scope
433        assert_eq!(tracker.health(&id4), ProjectionHealth::Healthy); // different kind
434    }
435
436    #[test]
437    fn invalidate_scope_affects_all_kinds() {
438        let mut tracker = ProjectionTracker::new(3600);
439        let scope = test_scope();
440
441        let id1 = ProjectionId::new(ProjectionKind::Entity, "a", scope.clone());
442        let id2 = ProjectionId::new(ProjectionKind::Temporal, "b", scope.clone());
443
444        tracker.record_build(id1.clone(), 5, 10, None);
445        tracker.record_build(id2.clone(), 5, 10, None);
446
447        let result = tracker.invalidate_scope(&scope, StaleCause::SourceChanged);
448        assert_eq!(result.affected_count, 2);
449        assert_eq!(tracker.health(&id1), ProjectionHealth::Stale);
450        assert_eq!(tracker.health(&id2), ProjectionHealth::Stale);
451    }
452
453    #[test]
454    fn query_status_filters() {
455        let mut tracker = ProjectionTracker::new(3600);
456        let scope = test_scope();
457
458        let id1 = ProjectionId::new(ProjectionKind::Entity, "a", scope.clone());
459        let id2 = ProjectionId::new(ProjectionKind::Temporal, "b", scope.clone());
460        tracker.record_build(id1.clone(), 5, 10, None);
461        tracker.record_build(id2.clone(), 5, 10, None);
462
463        let entities = tracker.query_status(Some(&ProjectionKind::Entity), None);
464        assert_eq!(entities.len(), 1);
465
466        let in_scope = tracker.query_status(None, Some(&scope));
467        assert_eq!(in_scope.len(), 2);
468    }
469
470    #[test]
471    fn failure_recorded() {
472        let mut tracker = ProjectionTracker::new(3600);
473        let id = test_pid("test");
474        tracker.record_failure(id.clone(), "timeout".into());
475        assert_eq!(tracker.health(&id), ProjectionHealth::Missing);
476        assert!(tracker.get(&id).unwrap().last_error.is_some());
477    }
478
479    #[test]
480    fn remove_forces_missing() {
481        let mut tracker = ProjectionTracker::new(3600);
482        let id = test_pid("test");
483        tracker.record_build(id.clone(), 5, 20, None);
484        tracker.remove(&id);
485        assert_eq!(tracker.health(&id), ProjectionHealth::Missing);
486    }
487
488    #[test]
489    fn clear_scope_removes_only_targeted() {
490        let mut tracker = ProjectionTracker::new(3600);
491        let scope_a = ScopeKey::namespace_only("a");
492        let scope_b = ScopeKey::namespace_only("b");
493
494        let id_a = ProjectionId::new(ProjectionKind::Entity, "x", scope_a.clone());
495        let id_b = ProjectionId::new(ProjectionKind::Entity, "y", scope_b.clone());
496
497        tracker.record_build(id_a.clone(), 5, 10, None);
498        tracker.record_build(id_b.clone(), 5, 10, None);
499
500        let result = tracker.clear_scope(&scope_a);
501        assert_eq!(result.affected_count, 1);
502        assert_eq!(tracker.health(&id_a), ProjectionHealth::Missing);
503        assert_eq!(tracker.health(&id_b), ProjectionHealth::Healthy);
504    }
505
506    #[test]
507    fn version_metadata_preserved() {
508        let mut tracker = ProjectionTracker::new(3600);
509        let id = test_pid("test");
510        let version = ProjectionVersion {
511            schema_version: Some("1.0".into()),
512            resolver_version: Some("2.0".into()),
513        };
514        tracker.record_build(id.clone(), 5, 10, Some(version.clone()));
515
516        let meta = tracker.get(&id).unwrap();
517        assert_eq!(
518            meta.version.as_ref().unwrap().schema_version,
519            Some("1.0".into())
520        );
521    }
522}
knowledge_runtime/projection/lifecycle.rs

knowledge_runtime/projection/
lifecycle.rs
