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    /// The decimal places the source stores or presents this channel at. Pairing writes it onto
25    /// the slot where none is declared, and the public API expresses served values at it. None
26    /// leaves the slot undeclared, which is served unrounded.
27    pub decimal_places: Option<i16>,
28}
29
30/// Asks a backend for readings for one stream since a cursor.
31#[derive(Debug, Clone)]
32pub struct StreamFetchRequest {
33    pub stream_id: Uuid,
34    pub source_key: String,
35    /// Last known reading time. None on a new stream or a full sync.
36    pub since: Option<DateTime<Utc>>,
37}
38
39/// A completeness claim over one stream: the readings sent alongside are the COMPLETE content of
40/// the source for this stream over `[from, to)`, read from `source_rows_read` source rows. The
41/// server diffs stored content against the payload and converges (new / changed / withdrawn);
42/// without a window the request is a bare append, exactly the old semantics.
43#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
44pub struct SourceWindow {
45    pub from: DateTime<Utc>,
46    pub to: DateTime<Utc>,
47    /// Source rows scanned to produce the payload. An empty payload over a window the store holds
48    /// readings for is refused server-side, so a decode failure cannot read as a source deletion.
49    pub source_rows_read: u64,
50    /// Instants the backend saw but could not carry (cell decode failures). The server retains
51    /// stored rows at these keys rather than withdrawing them.
52    #[serde(default, skip_serializing_if = "Vec::is_empty")]
53    pub dropped_times: Vec<DateTime<Utc>>,
54    /// Digest of the canonical payload, stamped by the driver before send. The server persists
55    /// it on a cleanly-applied pass and echoes it on the stream list, so the next cycle can skip
56    /// re-sending unchanged content. Opaque to the server; never computed server-side.
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    pub content_digest: Option<String>,
59}
60
61/// Readings fetched for one stream, ready to ingest.
62#[derive(Debug)]
63pub struct StreamReadings {
64    pub stream_id: Uuid,
65    pub source_key: String,
66    pub readings: Vec<IngestReading>,
67    /// Portal-precomputed mean/sd per replicate group, for server-side comparison.
68    pub audits: Vec<GroupAudit>,
69    /// Marks the readings as replicate collections; the API groups them per instant.
70    pub collection: bool,
71    /// The completeness claim, when this fetch read the source's full content for the stream.
72    pub window: Option<SourceWindow>,
73    /// Source-authored annotations riding this stream's payload (e.g. the standard curve the
74    /// source applied while producing a stored value). The driver registers them after the
75    /// stream's readings ingest; idempotent per (source_system, source_key).
76    pub annotations: Vec<AnnotationUpsert>,
77}
78
79impl StreamReadings {
80    /// Plain single-series readings: no audits, not a collection, no completeness claim.
81    pub fn new(stream_id: Uuid, source_key: String, readings: Vec<IngestReading>) -> Self {
82        Self {
83            stream_id,
84            source_key,
85            readings,
86            audits: Vec::new(),
87            collection: false,
88            window: None,
89            annotations: Vec::new(),
90        }
91    }
92}
93
94/// Status events fetched for one stream.
95#[derive(Debug)]
96pub struct StreamStatusEvents {
97    pub stream_id: Uuid,
98    pub source_key: String,
99    pub events: Vec<IngestStatusEvent>,
100}
101
102/// Everything the source holds that could become a stream, whether or not the connector takes it.
103///
104/// `discover_streams` reports what a connector accepted, so a column it declined and a group it has
105/// not discovered yet are invisible to every downstream check: no completeness window covers them,
106/// no receipt names them, and reconciliation cannot speak about them at all. This is what a sign-off
107/// against a retiring source has to read.
108///
109/// Taken channels are listed per group because that is the question being asked (is this station,
110/// whole, here); declined channels are listed once, source-wide, because a source declines a
111/// channel by its own rules and not per group.
112#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
113pub struct SourceInventory {
114    /// One entry per channel the connector carries, keyed as it registers.
115    pub candidates: Vec<SourceCandidate>,
116    /// Channels the source holds and the connector does not carry, with the connector's reason.
117    #[serde(default)]
118    pub declined: Vec<DeclinedChannel>,
119    /// Every group the source holds, so one with no channel at all is still named.
120    #[serde(default)]
121    pub groups: Vec<String>,
122}
123
124/// One channel the connector carries.
125#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
126pub struct SourceCandidate {
127    /// The key it registers under.
128    pub source_key: String,
129    /// The source's own grouping: the station, the location, whatever a report is read by.
130    #[serde(default, skip_serializing_if = "Option::is_none")]
131    pub group: Option<String>,
132}
133
134/// One channel the connector leaves behind, and why.
135#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
136pub struct DeclinedChannel {
137    /// The source's own name for it: a column, a location id.
138    pub channel: String,
139    pub reason: String,
140}
141
142impl SourceInventory {
143    /// The inventory a backend that declines nothing has: every descriptor it discovered.
144    #[must_use]
145    pub fn of_discovered(descriptors: &[StreamDescriptor]) -> Self {
146        let candidates: Vec<SourceCandidate> = descriptors
147            .iter()
148            .map(|d| SourceCandidate {
149                source_key: d.source_key.clone(),
150                group: d.source_path.split('/').nth(1).map(ToString::to_string),
151            })
152            .collect();
153        let mut groups: Vec<String> = candidates
154            .iter()
155            .filter_map(|c| c.group.clone())
156            .collect::<std::collections::BTreeSet<_>>()
157            .into_iter()
158            .collect();
159        groups.sort();
160        Self {
161            candidates,
162            declined: Vec::new(),
163            groups,
164        }
165    }
166}
river_data_core/models/backend.rs

river_data_core/models/
backend.rs
