river_data_core/models/
backend.rs

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

river_data_core/models/
backend.rs
