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, InstrumentGranularity};
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 /// The instrument each channel suggests at pairing: per parameter or per site and parameter.
29 pub instrument_granularity: Option<InstrumentGranularity>,
30}
31
32/// Asks a backend for readings for one stream since a cursor.
33#[derive(Debug, Clone)]
34pub struct StreamFetchRequest {
35 pub stream_id: Uuid,
36 pub source_key: String,
37 /// Last known reading time. None on a new stream or a full sync.
38 pub since: Option<DateTime<Utc>>,
39}
40
41/// A completeness claim over one stream: the readings sent alongside are the COMPLETE content of
42/// the source for this stream over `[from, to)`, read from `source_rows_read` source rows. The
43/// server diffs stored content against the payload and converges (new / changed / withdrawn);
44/// without a window the request is a bare append, exactly the old semantics.
45#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
46#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
47#[serde(deny_unknown_fields)]
48pub struct SourceWindow {
49 pub from: DateTime<Utc>,
50 pub to: DateTime<Utc>,
51 /// Source rows scanned to produce the payload. An empty payload over a window the store holds
52 /// readings for is refused server-side, so a decode failure cannot read as a source deletion.
53 pub source_rows_read: u64,
54 /// Instants the backend saw but could not carry (cell decode failures). The server retains
55 /// stored rows at these keys rather than withdrawing them.
56 #[serde(default, skip_serializing_if = "Vec::is_empty")]
57 pub dropped_times: Vec<DateTime<Utc>>,
58 /// Digest of the canonical payload, stamped by the driver before send. The server persists
59 /// it on a cleanly-applied pass and echoes it on the stream list, so the next cycle can skip
60 /// re-sending unchanged content. Opaque to the server; never computed server-side.
61 #[serde(default, skip_serializing_if = "Option::is_none")]
62 pub content_digest: Option<String>,
63}
64
65/// Readings fetched for one stream, ready to ingest.
66#[derive(Debug)]
67pub struct StreamReadings {
68 pub stream_id: Uuid,
69 pub source_key: String,
70 pub readings: Vec<IngestReading>,
71 /// Portal-precomputed mean/sd per replicate group, for server-side comparison.
72 pub audits: Vec<GroupAudit>,
73 /// Marks the readings as replicate collections; the API groups them per instant.
74 pub collection: bool,
75 /// The completeness claim, when this fetch read the source's full content for the stream.
76 pub window: Option<SourceWindow>,
77 /// Source-authored annotations riding this stream's payload (e.g. the standard curve the
78 /// source applied while producing a stored value). The driver registers them after the
79 /// stream's readings ingest; idempotent per (source_system, source_key).
80 pub annotations: Vec<AnnotationUpsert>,
81}
82
83impl StreamReadings {
84 /// Plain single-series readings: no audits, not a collection, no completeness claim.
85 pub fn new(stream_id: Uuid, source_key: String, readings: Vec<IngestReading>) -> Self {
86 Self {
87 stream_id,
88 source_key,
89 readings,
90 audits: Vec::new(),
91 collection: false,
92 window: None,
93 annotations: Vec::new(),
94 }
95 }
96}
97
98/// Status events fetched for one stream.
99#[derive(Debug)]
100pub struct StreamStatusEvents {
101 pub stream_id: Uuid,
102 pub source_key: String,
103 pub events: Vec<IngestStatusEvent>,
104}
105
106/// Everything the source holds that could become a stream, whether or not the connector takes it.
107///
108/// `discover_streams` reports what a connector accepted, so a column it declined and a group it has
109/// not discovered yet are invisible to every downstream check: no completeness window covers them,
110/// no receipt names them, and reconciliation cannot speak about them at all. This is what a sign-off
111/// against a retiring source has to read.
112///
113/// Taken channels are listed per group because that is the question being asked (is this station,
114/// whole, here); declined channels are listed once, source-wide, because a source declines a
115/// channel by its own rules and not per group.
116#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
117pub struct SourceInventory {
118 /// One entry per channel the connector carries, keyed as it registers.
119 pub candidates: Vec<SourceCandidate>,
120 /// Channels the source holds and the connector does not carry, with the connector's reason.
121 #[serde(default)]
122 pub declined: Vec<DeclinedChannel>,
123 /// Every group the source holds, so one with no channel at all is still named.
124 #[serde(default)]
125 pub groups: Vec<String>,
126 /// The source's own instrument register, where it keeps one. These are instruments no stream
127 /// mints: a portal's `sensor_inventory` is the answer to which probe, which serial, installed
128 /// when, and it goes with the portal unless it is admitted here.
129 #[serde(default)]
130 pub instruments: Vec<crate::models::SensorUpsert>,
131}
132
133/// One channel the connector carries.
134#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
135pub struct SourceCandidate {
136 /// The key it registers under.
137 pub source_key: String,
138 /// The source's own grouping: the station, the location, whatever a report is read by.
139 #[serde(default, skip_serializing_if = "Option::is_none")]
140 pub group: Option<String>,
141}
142
143/// One channel the connector leaves behind, and why.
144#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
145pub struct DeclinedChannel {
146 /// The source's own name for it: a column, a location id.
147 pub channel: String,
148 pub reason: String,
149}
150
151impl SourceInventory {
152 /// The inventory a backend that declines nothing has: every descriptor it discovered.
153 #[must_use]
154 pub fn of_discovered(descriptors: &[StreamDescriptor]) -> Self {
155 let candidates: Vec<SourceCandidate> = descriptors
156 .iter()
157 .map(|d| SourceCandidate {
158 source_key: d.source_key.clone(),
159 group: d.source_path.split('/').nth(1).map(ToString::to_string),
160 })
161 .collect();
162 let mut groups: Vec<String> = candidates
163 .iter()
164 .filter_map(|c| c.group.clone())
165 .collect::<std::collections::BTreeSet<_>>()
166 .into_iter()
167 .collect();
168 groups.sort();
169 Self {
170 candidates,
171 declined: Vec::new(),
172 groups,
173 instruments: Vec::new(),
174 }
175 }
176}
177
178#[cfg(test)]
179mod tests {
180 use super::*;
181
182 /// The completeness claim round-trips: the API echoes it back as `accepted_window` and the
183 /// client treats a missing echo as a hard error, so both sides read one shape.
184 #[test]
185 fn source_window_round_trips() {
186 let w = SourceWindow {
187 from: Utc::now(),
188 to: Utc::now(),
189 source_rows_read: 500,
190 dropped_times: vec![Utc::now()],
191 content_digest: Some("fnv:1".into()),
192 };
193 let back: SourceWindow = serde_json::from_value(serde_json::to_value(&w).unwrap()).unwrap();
194 assert_eq!(back.source_rows_read, 500);
195 assert_eq!(back.dropped_times.len(), 1);
196 assert_eq!(back.content_digest.as_deref(), Some("fnv:1"));
197 }
198}