river_data_core/models/
backend.rs

1use chrono::{DateTime, Utc};
2use uuid::Uuid;
3
4use crate::models::annotations::AnnotationUpsert;
5use crate::models::replicates::{GroupAudit, ReplicateSpec};
6use crate::models::streams::{IngestReading, IngestStatusEvent};
7
8/// Describes a data stream to register with river-data.
9#[derive(Debug, Clone)]
10pub struct StreamDescriptor {
11    /// Unique key within the source system (ie. a location id or column name).
12    pub source_key: String,
13    /// Human-readable name shown in the dashboard.
14    pub source_name: String,
15    /// Hierarchy path (ie. "cnet/VAD/WTW_DO_mgL_1"), parsed server-side for site discovery.
16    pub source_path: String,
17    pub metadata: serde_json::Value,
18    /// Stream classification ('spot' or 'continuous'); None defers to the API's resolution chain.
19    pub measurement_type: Option<String>,
20    /// Owning sensor; required for streams whose readings carry curve claims.
21    pub sensor_id: Option<Uuid>,
22    /// Replicate-family declaration; requires `measurement_type: "spot"`.
23    pub replicates: Option<ReplicateSpec>,
24}
25
26/// Asks a backend for readings for one stream since a cursor.
27#[derive(Debug, Clone)]
28pub struct StreamFetchRequest {
29    pub stream_id: Uuid,
30    pub source_key: String,
31    /// Last known reading time. None on a new stream or a full sync.
32    pub since: Option<DateTime<Utc>>,
33}
34
35/// A completeness claim over one stream: the readings sent alongside are the COMPLETE content of
36/// the source for this stream over `[from, to)`, read from `source_rows_read` source rows. The
37/// server diffs stored content against the payload and converges (new / changed / withdrawn);
38/// without a window the request is a bare append, exactly the old semantics.
39#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
40pub struct SourceWindow {
41    pub from: DateTime<Utc>,
42    pub to: DateTime<Utc>,
43    /// Source rows scanned to produce the payload. An empty payload over a window the store holds
44    /// readings for is refused server-side, so a decode failure cannot read as a source deletion.
45    pub source_rows_read: u64,
46    /// Instants the backend saw but could not carry (cell decode failures). The server retains
47    /// stored rows at these keys rather than withdrawing them.
48    #[serde(default, skip_serializing_if = "Vec::is_empty")]
49    pub dropped_times: Vec<DateTime<Utc>>,
50    /// Digest of the canonical payload, stamped by the driver before send. The server persists
51    /// it on a cleanly-applied pass and echoes it on the stream list, so the next cycle can skip
52    /// re-sending unchanged content. Opaque to the server; never computed server-side.
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub content_digest: Option<String>,
55}
56
57/// Readings fetched for one stream, ready to ingest.
58#[derive(Debug)]
59pub struct StreamReadings {
60    pub stream_id: Uuid,
61    pub source_key: String,
62    pub readings: Vec<IngestReading>,
63    /// Portal-precomputed mean/sd per replicate group, for server-side comparison.
64    pub audits: Vec<GroupAudit>,
65    /// Marks the readings as replicate collections; the API groups them per instant.
66    pub collection: bool,
67    /// The completeness claim, when this fetch read the source's full content for the stream.
68    pub window: Option<SourceWindow>,
69    /// Source-authored annotations riding this stream's payload (e.g. the standard curve the
70    /// source applied while producing a stored value). The driver registers them after the
71    /// stream's readings ingest; idempotent per (source_system, source_key).
72    pub annotations: Vec<AnnotationUpsert>,
73}
74
75impl StreamReadings {
76    /// Plain single-series readings: no audits, not a collection, no completeness claim.
77    pub fn new(stream_id: Uuid, source_key: String, readings: Vec<IngestReading>) -> Self {
78        Self {
79            stream_id,
80            source_key,
81            readings,
82            audits: Vec::new(),
83            collection: false,
84            window: None,
85            annotations: Vec::new(),
86        }
87    }
88}
89
90/// Status events fetched for one stream.
91#[derive(Debug)]
92pub struct StreamStatusEvents {
93    pub stream_id: Uuid,
94    pub source_key: String,
95    pub events: Vec<IngestStatusEvent>,
96}
river_data_core/models/backend.rs

river_data_core/models/
backend.rs
