river_data_core/
models.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4// ============================================================================
5// Status Enums
6// ============================================================================
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum ServiceStatus {
11    Starting,
12    Idle,
13    Running,
14    Paused,
15    Syncing,
16    Error,
17    Stopping,
18}
19
20impl ServiceStatus {
21    pub const ALL: &[ServiceStatus] = &[
22        Self::Starting, Self::Idle, Self::Running, Self::Paused,
23        Self::Syncing, Self::Error, Self::Stopping,
24    ];
25
26    pub fn as_str(&self) -> &'static str {
27        match self {
28            Self::Starting => "starting",
29            Self::Idle => "idle",
30            Self::Running => "running",
31            Self::Paused => "paused",
32            Self::Syncing => "syncing",
33            Self::Error => "error",
34            Self::Stopping => "stopping",
35        }
36    }
37
38    pub fn from_str(s: &str) -> Option<Self> {
39        Self::ALL.iter().find(|v| v.as_str() == s).copied()
40    }
41}
42
43impl std::fmt::Display for ServiceStatus {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        f.write_str(self.as_str())
46    }
47}
48
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum CommandStatus {
52    Pending,
53    Acknowledged,
54    Completed,
55    Failed,
56    Expired,
57}
58
59impl CommandStatus {
60    pub const fn as_str(&self) -> &'static str {
61        match self {
62            Self::Pending => "pending",
63            Self::Acknowledged => "acknowledged",
64            Self::Completed => "completed",
65            Self::Failed => "failed",
66            Self::Expired => "expired",
67        }
68    }
69}
70
71impl std::fmt::Display for CommandStatus {
72    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73        f.write_str(self.as_str())
74    }
75}
76
77#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
78#[serde(rename_all = "snake_case")]
79pub enum SyncEventType {
80    Scheduled,
81    Manual,
82    Command,
83    Triggered,
84    FullSync,
85}
86
87impl SyncEventType {
88    pub const ALL: &[SyncEventType] = &[
89        Self::Scheduled, Self::Manual, Self::Command, Self::Triggered, Self::FullSync,
90    ];
91
92    pub fn as_str(&self) -> &'static str {
93        match self {
94            Self::Scheduled => "scheduled",
95            Self::Manual => "manual",
96            Self::Command => "command",
97            Self::Triggered => "triggered",
98            Self::FullSync => "full_sync",
99        }
100    }
101
102    pub fn from_str(s: &str) -> Option<Self> {
103        Self::ALL.iter().find(|v| v.as_str() == s).copied()
104    }
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108#[serde(rename_all = "snake_case")]
109pub enum SyncEventStatus {
110    Running,
111    Completed,
112    Partial,
113    Failed,
114    Cancelled,
115}
116
117impl SyncEventStatus {
118    pub const ALL: &[SyncEventStatus] = &[
119        Self::Running, Self::Completed, Self::Partial, Self::Failed, Self::Cancelled,
120    ];
121
122    pub fn as_str(&self) -> &'static str {
123        match self {
124            Self::Running => "running",
125            Self::Completed => "completed",
126            Self::Partial => "partial",
127            Self::Failed => "failed",
128            Self::Cancelled => "cancelled",
129        }
130    }
131
132    pub fn from_str(s: &str) -> Option<Self> {
133        Self::ALL.iter().find(|v| v.as_str() == s).copied()
134    }
135
136    pub fn is_terminal(&self) -> bool {
137        matches!(self, Self::Completed | Self::Partial | Self::Failed | Self::Cancelled)
138    }
139
140    pub fn is_success(&self) -> bool {
141        matches!(self, Self::Completed | Self::Partial)
142    }
143}
144
145// ============================================================================
146// Server Configuration
147// ============================================================================
148
149#[derive(Debug, Clone)]
150pub struct SyncServerConfig {
151    pub session_token_ttl_secs: u64,
152    pub token_cache_capacity: u64,
153    pub token_cache_ttl_secs: u64,
154    pub command_expiry_secs: u64,
155    pub health_healthy_secs: i64,
156    pub health_warning_secs: i64,
157    pub client_id_prefix: String,
158}
159
160impl Default for SyncServerConfig {
161    fn default() -> Self {
162        Self {
163            session_token_ttl_secs: 900,
164            token_cache_capacity: 100,
165            token_cache_ttl_secs: 780,
166            command_expiry_secs: 300,
167            health_healthy_secs: 90,
168            health_warning_secs: 300,
169            client_id_prefix: "svc_".to_string(),
170        }
171    }
172}
173
174// ============================================================================
175// Enrollment
176// ============================================================================
177
178#[derive(Debug, Serialize, Deserialize)]
179#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
180pub struct EnrollRequest {
181    pub client_id: String,
182    pub client_secret: String,
183    pub instance_id: String,
184}
185
186#[derive(Debug, Serialize, Deserialize)]
187#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
188pub struct EnrollResponse {
189    pub service_id: Uuid,
190    pub session_token: String,
191}
192
193// ============================================================================
194// Heartbeat
195// ============================================================================
196
197#[derive(Debug, Serialize, Deserialize)]
198#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
199pub struct HeartbeatRequest {
200    pub service_id: Uuid,
201    pub status: String,
202    pub current_operation: Option<String>,
203}
204
205#[derive(Debug, Serialize, Deserialize)]
206#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
207pub struct HeartbeatResponse {
208    pub session_token: String,
209    pub pending_commands: Vec<PendingCommand>,
210}
211
212#[derive(Debug, Clone, Serialize, Deserialize)]
213#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
214pub struct PendingCommand {
215    pub id: Uuid,
216    pub command: String,
217    #[cfg_attr(feature = "server", schema(value_type = Object))]
218    pub payload: Option<serde_json::Value>,
219}
220
221// ============================================================================
222// Command Updates
223// ============================================================================
224
225#[derive(Debug, Serialize, Deserialize)]
226#[cfg_attr(feature = "server", derive(utoipa::ToSchema))]
227pub struct CommandUpdateRequest {
228    pub status: String,
229    #[cfg_attr(feature = "server", schema(value_type = Object))]
230    pub result: Option<serde_json::Value>,
231}
232
233// ============================================================================
234// Sync Result
235// ============================================================================
236
237#[derive(Debug, Default, Serialize)]
238pub struct SyncResult {
239    pub readings_synced: u64,
240    pub status_events_synced: u64,
241    pub full_sync: bool,
242    pub duration_ms: u64,
243    #[serde(skip_serializing_if = "Vec::is_empty")]
244    pub errors: Vec<String>,
245    #[serde(skip_serializing_if = "Vec::is_empty")]
246    pub log: Vec<String>,
247}
248
249#[derive(Debug)]
250pub enum SyncTrigger {
251    Scheduled,
252    Command { id: Uuid, full: bool },
253}
254
255// ============================================================================
256// Runner Config
257// ============================================================================
258
259#[derive(Debug, Clone)]
260pub struct RunnerConfig {
261    pub api_base_url: String,
262    pub client_id: String,
263    pub client_secret: String,
264    pub instance_id: String,
265    pub heartbeat_interval_secs: u64,
266    pub sync_interval_secs: u64,
267    pub enrollment_retry_secs: u64,
268}
269
270impl RunnerConfig {
271    pub fn from_env() -> Result<Self, String> {
272        Ok(Self {
273            api_base_url: require_env("API_BASE_URL")?,
274            client_id: require_env("SERVICE_CLIENT_ID")?,
275            client_secret: require_env("SERVICE_CLIENT_SECRET")?,
276            instance_id: std::env::var("INSTANCE_ID").unwrap_or_else(|_| "default".to_string()),
277            heartbeat_interval_secs: env_u64("HEARTBEAT_INTERVAL_SECONDS", 30),
278            sync_interval_secs: env_u64("SYNC_INTERVAL_SECONDS", 300),
279            enrollment_retry_secs: env_u64("ENROLLMENT_RETRY_SECONDS", 10),
280        })
281    }
282}
283
284fn require_env(key: &str) -> Result<String, String> {
285    std::env::var(key).map_err(|_| format!("Missing required env var: {key}"))
286}
287
288fn env_u64(key: &str, default: u64) -> u64 {
289    std::env::var(key)
290        .ok()
291        .and_then(|v| v.parse().ok())
292        .unwrap_or(default)
293}
294
295// ============================================================================
296// River Data API types (shared across sync services)
297// ============================================================================
298
299#[derive(Debug, Clone, Serialize, Deserialize)]
300pub struct DataStream {
301    pub id: Uuid,
302    pub source_system: String,
303    pub source_key: String,
304    pub source_name: Option<String>,
305    pub source_path: Option<String>,
306    pub metadata: serde_json::Value,
307    pub site_parameter_id: Option<Uuid>,
308    /// Stream-level default for readings.measurement_type ('continuous' | 'spot' | 'derived').
309    /// None defers to the API's sensor-frequency resolution.
310    #[serde(default, skip_serializing_if = "Option::is_none")]
311    pub measurement_type: Option<String>,
312    pub is_active: bool,
313    pub last_data_time: Option<chrono::DateTime<chrono::Utc>>,
314}
315
316#[derive(Debug, Serialize)]
317pub struct RegisterStreamRequest {
318    pub source_system: String,
319    pub source_key: String,
320    pub source_name: Option<String>,
321    pub source_path: Option<String>,
322    pub metadata: serde_json::Value,
323    /// Stream-level classification declared at discovery. None never clears an operator-set value.
324    #[serde(skip_serializing_if = "Option::is_none")]
325    pub measurement_type: Option<String>,
326}
327
328#[derive(Debug, Clone, Serialize)]
329pub struct IngestReading {
330    pub time: chrono::DateTime<chrono::Utc>,
331    pub raw_value: f64,
332    #[serde(skip_serializing_if = "is_zero")]
333    pub replicate_index: i16,
334    #[serde(skip_serializing_if = "Option::is_none")]
335    pub sensor_id: Option<Uuid>,
336    #[serde(skip_serializing_if = "Option::is_none")]
337    pub calibration_id: Option<Uuid>,
338    #[serde(skip_serializing_if = "Option::is_none")]
339    pub deployment_id: Option<Uuid>,
340    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
341    /// the stream default, then the owning sensor's data_frequency.
342    #[serde(skip_serializing_if = "Option::is_none")]
343    pub measurement_type: Option<String>,
344}
345
346fn is_zero(v: &i16) -> bool {
347    *v == 0
348}
349
350#[derive(Debug, Serialize)]
351pub struct IngestStatusEvent {
352    pub time: chrono::DateTime<chrono::Utc>,
353    pub value: String,
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359
360    #[test]
361    fn test_enroll_request_serialization() {
362        let req = EnrollRequest {
363            client_id: "svc_abc".to_string(),
364            client_secret: "secret123".to_string(),
365            instance_id: "service-01".to_string(),
366        };
367        let json = serde_json::to_value(&req).unwrap();
368        assert_eq!(json["client_id"], "svc_abc");
369        assert_eq!(json["instance_id"], "service-01");
370    }
371
372    #[test]
373    fn test_enroll_response_deserialization() {
374        let json = serde_json::json!({
375            "service_id": "550e8400-e29b-41d4-a716-446655440000",
376            "session_token": "tok-abc"
377        });
378        let resp: EnrollResponse = serde_json::from_value(json).unwrap();
379        assert_eq!(resp.session_token, "tok-abc");
380    }
381
382    #[test]
383    fn test_heartbeat_response_with_commands() {
384        let json = serde_json::json!({
385            "session_token": "new-tok",
386            "pending_commands": [
387                {
388                    "id": "550e8400-e29b-41d4-a716-446655440000",
389                    "command": "trigger_sync",
390                    "payload": null
391                }
392            ]
393        });
394        let resp: HeartbeatResponse = serde_json::from_value(json).unwrap();
395        assert_eq!(resp.pending_commands.len(), 1);
396        assert_eq!(resp.pending_commands[0].command, "trigger_sync");
397    }
398
399    #[test]
400    fn test_sync_result_default() {
401        let r = SyncResult::default();
402        assert_eq!(r.readings_synced, 0);
403        assert!(!r.full_sync);
404        assert!(r.errors.is_empty());
405    }
406
407    #[test]
408    fn test_sync_result_serialization_skips_empty() {
409        let r = SyncResult {
410            readings_synced: 100,
411            ..Default::default()
412        };
413        let json = serde_json::to_value(&r).unwrap();
414        assert_eq!(json["readings_synced"], 100);
415        assert!(json.get("errors").is_none());
416    }
417
418    #[test]
419    fn test_ingest_reading_serialization() {
420        let r = IngestReading {
421            time: chrono::Utc::now(),
422            raw_value: 42.5,
423            replicate_index: 0,
424            sensor_id: None,
425            calibration_id: None,
426            deployment_id: None,
427            measurement_type: None,
428        };
429        let json = serde_json::to_value(&r).unwrap();
430        assert_eq!(json["raw_value"], 42.5);
431        assert!(json.get("replicate_index").is_none());
432        assert!(json.get("sensor_id").is_none());
433        assert!(json.get("measurement_type").is_none());
434    }
435
436    #[test]
437    fn test_register_stream_request() {
438        let req = RegisterStreamRequest {
439            source_system: "test_system".to_string(),
440            source_key: "source_1".to_string(),
441            source_name: Some("stream_a".to_string()),
442            source_path: None,
443            metadata: serde_json::json!({"device": "dev_001"}),
444            measurement_type: None,
445        };
446        let json = serde_json::to_value(&req).unwrap();
447        assert_eq!(json["source_system"], "test_system");
448        assert_eq!(json["metadata"]["device"], "dev_001");
449    }
450
451    #[test]
452    fn test_data_stream_deserialization() {
453        let json = serde_json::json!({
454            "id": "550e8400-e29b-41d4-a716-446655440000",
455            "source_system": "test_system",
456            "source_key": "source_1",
457            "source_name": "stream_a",
458            "source_path": null,
459            "metadata": {},
460            "site_parameter_id": null,
461            "is_active": true,
462            "last_data_time": null
463        });
464        let stream: DataStream = serde_json::from_value(json).unwrap();
465        assert_eq!(stream.source_system, "test_system");
466        assert!(stream.is_active);
467        assert!(stream.site_parameter_id.is_none());
468    }
469}
river_data_core/models.rs

river_data_core/
models.rs
