river_data_core/
models.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4// ============================================================================
5// Enrollment
6// ============================================================================
7
8#[derive(Debug, Serialize)]
9pub struct EnrollRequest {
10    pub client_id: String,
11    pub client_secret: String,
12    pub instance_id: String,
13}
14
15#[derive(Debug, Deserialize)]
16pub struct EnrollResponse {
17    pub service_id: Uuid,
18    pub session_token: String,
19}
20
21// ============================================================================
22// Heartbeat
23// ============================================================================
24
25#[derive(Debug, Serialize)]
26pub struct HeartbeatRequest {
27    pub service_id: Uuid,
28    pub status: String,
29    pub current_operation: Option<String>,
30}
31
32#[derive(Debug, Deserialize)]
33pub struct HeartbeatResponse {
34    pub session_token: String,
35    pub pending_commands: Vec<PendingCommand>,
36}
37
38#[derive(Debug, Clone, Deserialize)]
39pub struct PendingCommand {
40    pub id: Uuid,
41    pub command: String,
42    pub payload: Option<serde_json::Value>,
43}
44
45// ============================================================================
46// Command Updates
47// ============================================================================
48
49#[derive(Debug, Serialize)]
50pub struct CommandUpdateRequest {
51    pub status: String,
52    pub result: Option<serde_json::Value>,
53}
54
55// ============================================================================
56// Sync Result
57// ============================================================================
58
59#[derive(Debug, Default, Serialize)]
60pub struct SyncResult {
61    pub readings_synced: u64,
62    pub status_events_synced: u64,
63    pub full_sync: bool,
64    pub duration_ms: u64,
65    #[serde(skip_serializing_if = "Vec::is_empty")]
66    pub errors: Vec<String>,
67    #[serde(skip_serializing_if = "Vec::is_empty")]
68    pub log: Vec<String>,
69}
70
71#[derive(Debug)]
72pub enum SyncTrigger {
73    Scheduled,
74    Command { id: Uuid, full: bool },
75}
76
77// ============================================================================
78// Runner Config
79// ============================================================================
80
81#[derive(Debug, Clone)]
82pub struct RunnerConfig {
83    pub api_base_url: String,
84    pub client_id: String,
85    pub client_secret: String,
86    pub instance_id: String,
87    pub heartbeat_interval_secs: u64,
88    pub sync_interval_secs: u64,
89}
90
91impl RunnerConfig {
92    pub fn from_env() -> Result<Self, String> {
93        Ok(Self {
94            api_base_url: require_env("API_BASE_URL")?,
95            client_id: require_env("SERVICE_CLIENT_ID")?,
96            client_secret: require_env("SERVICE_CLIENT_SECRET")?,
97            instance_id: std::env::var("INSTANCE_ID").unwrap_or_else(|_| "default".to_string()),
98            heartbeat_interval_secs: env_u64("HEARTBEAT_INTERVAL_SECONDS", 30),
99            sync_interval_secs: env_u64("SYNC_INTERVAL_SECONDS", 300),
100        })
101    }
102}
103
104fn require_env(key: &str) -> Result<String, String> {
105    std::env::var(key).map_err(|_| format!("Missing required env var: {key}"))
106}
107
108fn env_u64(key: &str, default: u64) -> u64 {
109    std::env::var(key)
110        .ok()
111        .and_then(|v| v.parse().ok())
112        .unwrap_or(default)
113}
114
115// ============================================================================
116// River Data API types (shared across sync services)
117// ============================================================================
118
119#[derive(Debug, Clone, Serialize, Deserialize)]
120pub struct DataStream {
121    pub id: Uuid,
122    pub source_system: String,
123    pub source_key: String,
124    pub source_name: Option<String>,
125    pub source_path: Option<String>,
126    pub metadata: serde_json::Value,
127    pub site_parameter_id: Option<Uuid>,
128    pub is_active: bool,
129    pub last_data_time: Option<chrono::DateTime<chrono::Utc>>,
130}
131
132#[derive(Debug, Serialize)]
133pub struct RegisterStreamRequest {
134    pub source_system: String,
135    pub source_key: String,
136    pub source_name: Option<String>,
137    pub source_path: Option<String>,
138    pub metadata: serde_json::Value,
139}
140
141#[derive(Debug, Serialize)]
142pub struct IngestReadingsRequest {
143    pub stream_id: Uuid,
144    pub readings: Vec<IngestReading>,
145}
146
147#[derive(Debug, Clone, Serialize)]
148pub struct IngestReading {
149    pub time: chrono::DateTime<chrono::Utc>,
150    pub raw_value: f64,
151    #[serde(skip_serializing_if = "is_zero")]
152    pub replicate_index: i16,
153    #[serde(skip_serializing_if = "Option::is_none")]
154    pub sensor_id: Option<Uuid>,
155    #[serde(skip_serializing_if = "Option::is_none")]
156    pub calibration_id: Option<Uuid>,
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub deployment_id: Option<Uuid>,
159}
160
161fn is_zero(v: &i16) -> bool {
162    *v == 0
163}
164
165#[derive(Debug, Serialize)]
166pub struct IngestStatusEventsRequest {
167    pub stream_id: Uuid,
168    pub events: Vec<IngestStatusEvent>,
169}
170
171#[derive(Debug, Serialize)]
172pub struct IngestStatusEvent {
173    pub time: chrono::DateTime<chrono::Utc>,
174    pub value: String,
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180
181    #[test]
182    fn test_enroll_request_serialization() {
183        let req = EnrollRequest {
184            client_id: "svc_abc".to_string(),
185            client_secret: "secret123".to_string(),
186            instance_id: "vaisala-01".to_string(),
187        };
188        let json = serde_json::to_value(&req).unwrap();
189        assert_eq!(json["client_id"], "svc_abc");
190        assert_eq!(json["instance_id"], "vaisala-01");
191    }
192
193    #[test]
194    fn test_enroll_response_deserialization() {
195        let json = serde_json::json!({
196            "service_id": "550e8400-e29b-41d4-a716-446655440000",
197            "session_token": "tok-abc"
198        });
199        let resp: EnrollResponse = serde_json::from_value(json).unwrap();
200        assert_eq!(resp.session_token, "tok-abc");
201    }
202
203    #[test]
204    fn test_heartbeat_response_with_commands() {
205        let json = serde_json::json!({
206            "session_token": "new-tok",
207            "pending_commands": [
208                {
209                    "id": "550e8400-e29b-41d4-a716-446655440000",
210                    "command": "trigger_sync",
211                    "payload": null
212                }
213            ]
214        });
215        let resp: HeartbeatResponse = serde_json::from_value(json).unwrap();
216        assert_eq!(resp.pending_commands.len(), 1);
217        assert_eq!(resp.pending_commands[0].command, "trigger_sync");
218    }
219
220    #[test]
221    fn test_sync_result_default() {
222        let r = SyncResult::default();
223        assert_eq!(r.readings_synced, 0);
224        assert!(!r.full_sync);
225        assert!(r.errors.is_empty());
226    }
227
228    #[test]
229    fn test_sync_result_serialization_skips_empty() {
230        let r = SyncResult {
231            readings_synced: 100,
232            ..Default::default()
233        };
234        let json = serde_json::to_value(&r).unwrap();
235        assert_eq!(json["readings_synced"], 100);
236        assert!(json.get("errors").is_none());
237    }
238
239    #[test]
240    fn test_ingest_reading_serialization() {
241        let r = IngestReading {
242            time: chrono::Utc::now(),
243            raw_value: 42.5,
244            replicate_index: 0,
245            sensor_id: None,
246            calibration_id: None,
247            deployment_id: None,
248        };
249        let json = serde_json::to_value(&r).unwrap();
250        assert_eq!(json["raw_value"], 42.5);
251        assert!(json.get("replicate_index").is_none());
252        assert!(json.get("sensor_id").is_none());
253    }
254
255    #[test]
256    fn test_register_stream_request() {
257        let req = RegisterStreamRequest {
258            source_system: "vaisala".to_string(),
259            source_key: "loc_1270".to_string(),
260            source_name: Some("MDepthmm".to_string()),
261            source_path: None,
262            metadata: serde_json::json!({"device": "25284027"}),
263        };
264        let json = serde_json::to_value(&req).unwrap();
265        assert_eq!(json["source_system"], "vaisala");
266        assert_eq!(json["metadata"]["device"], "25284027");
267    }
268
269    #[test]
270    fn test_data_stream_deserialization() {
271        let json = serde_json::json!({
272            "id": "550e8400-e29b-41d4-a716-446655440000",
273            "source_system": "vaisala",
274            "source_key": "loc_1270",
275            "source_name": "MDepthmm",
276            "source_path": null,
277            "metadata": {},
278            "site_parameter_id": null,
279            "is_active": true,
280            "last_data_time": null
281        });
282        let stream: DataStream = serde_json::from_value(json).unwrap();
283        assert_eq!(stream.source_system, "vaisala");
284        assert!(stream.is_active);
285        assert!(stream.site_parameter_id.is_none());
286    }
287}
river_data_core/models.rs

river_data_core/
models.rs
