river_data_core/models/
protocol.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4use crate::models::status::{SyncEventStatus, SyncEventType};
5
6#[derive(Debug, Serialize, Deserialize)]
7#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
8pub struct EnrollRequest {
9    pub client_id: String,
10    pub client_secret: String,
11    pub instance_id: String,
12}
13
14#[derive(Debug, Serialize, Deserialize)]
15#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
16pub struct EnrollResponse {
17    pub service_id: Uuid,
18    pub session_token: String,
19    /// Operator-desired pause state, persisted server-side; honored before the
20    /// initial sync so a restart cannot undo a pause.
21    #[serde(default)]
22    pub paused: bool,
23    /// Operator-set scheduled sync cadence, persisted server-side. None leaves the
24    /// service on its own `SYNC_INTERVAL_SECONDS`.
25    #[serde(default)]
26    pub sync_interval_secs: Option<u64>,
27}
28
29#[derive(Debug, Serialize, Deserialize)]
30#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
31pub struct HeartbeatRequest {
32    pub service_id: Uuid,
33    pub status: String,
34    pub current_operation: Option<String>,
35}
36
37#[derive(Debug, Serialize, Deserialize)]
38#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
39pub struct HeartbeatResponse {
40    pub session_token: String,
41    pub pending_commands: Vec<PendingCommand>,
42    /// Operator-desired pause state, persisted server-side.
43    #[serde(default)]
44    pub paused: bool,
45    /// Operator-set scheduled sync cadence, persisted server-side. A change here is
46    /// adopted by the running service on the next heartbeat.
47    #[serde(default)]
48    pub sync_interval_secs: Option<u64>,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize)]
52#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
53pub struct PendingCommand {
54    pub id: Uuid,
55    pub command: String,
56    #[cfg_attr(feature = "openapi", schema(value_type = Object))]
57    pub payload: Option<serde_json::Value>,
58}
59
60#[derive(Debug, Serialize, Deserialize)]
61#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
62pub struct CommandUpdateRequest {
63    pub status: String,
64    #[cfg_attr(feature = "openapi", schema(value_type = Object))]
65    pub result: Option<serde_json::Value>,
66}
67
68#[derive(Debug, Serialize)]
69pub struct SyncEventCreate {
70    pub service_id: Uuid,
71    #[serde(skip_serializing_if = "Option::is_none")]
72    pub command_id: Option<Uuid>,
73    pub event_type: SyncEventType,
74    pub status: SyncEventStatus,
75}
76
77#[derive(Debug, Deserialize)]
78pub struct SyncEventRef {
79    pub id: Uuid,
80}
81
82#[derive(Debug, Default, Serialize)]
83pub struct SyncEventUpdate {
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub status: Option<SyncEventStatus>,
86    #[serde(skip_serializing_if = "Option::is_none")]
87    pub readings_synced: Option<u64>,
88    /// Readings the API refused admission. Carried on the event rather than left to the process
89    /// log, so a stream losing rows every cycle leaves a queryable trace.
90    #[serde(skip_serializing_if = "Option::is_none")]
91    pub readings_skipped: Option<u64>,
92    #[serde(skip_serializing_if = "Option::is_none")]
93    pub status_events_synced: Option<u64>,
94    #[serde(skip_serializing_if = "Vec::is_empty")]
95    pub errors: Vec<String>,
96    #[serde(skip_serializing_if = "Vec::is_empty")]
97    pub log: Vec<String>,
98    #[serde(skip_serializing_if = "Option::is_none")]
99    pub duration_ms: Option<u64>,
100}
101
102/// Outcome of one sync cycle. The runner fills `full_sync` and `duration_ms`;
103/// a `SyncService` only reports counts, errors and log lines.
104#[derive(Debug, Default, Serialize)]
105pub struct SyncResult {
106    pub readings_synced: u64,
107    /// Readings the API refused admission and dropped. Additive: a reader that
108    /// predates the field must still parse the rest.
109    #[serde(default)]
110    pub readings_skipped: u64,
111    pub status_events_synced: u64,
112    pub full_sync: bool,
113    pub duration_ms: u64,
114    #[serde(skip_serializing_if = "Vec::is_empty")]
115    pub errors: Vec<String>,
116    #[serde(skip_serializing_if = "Vec::is_empty")]
117    pub log: Vec<String>,
118}
119
120#[derive(Debug)]
121pub enum SyncTrigger {
122    Scheduled,
123    Command { id: Uuid, full: bool },
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[test]
131    fn test_enroll_request_serialization() {
132        let req = EnrollRequest {
133            client_id: "svc_abc".to_string(),
134            client_secret: "secret123".to_string(),
135            instance_id: "service-01".to_string(),
136        };
137        let json = serde_json::to_value(&req).unwrap();
138        assert_eq!(json["client_id"], "svc_abc");
139        assert_eq!(json["instance_id"], "service-01");
140    }
141
142    #[test]
143    fn test_enroll_response_deserialization() {
144        let json = serde_json::json!({
145            "service_id": "550e8400-e29b-41d4-a716-446655440000",
146            "session_token": "tok-abc"
147        });
148        let resp: EnrollResponse = serde_json::from_value(json).unwrap();
149        assert_eq!(resp.session_token, "tok-abc");
150    }
151
152    #[test]
153    fn test_heartbeat_response_with_commands() {
154        let json = serde_json::json!({
155            "session_token": "new-tok",
156            "pending_commands": [
157                {
158                    "id": "550e8400-e29b-41d4-a716-446655440000",
159                    "command": "trigger_sync",
160                    "payload": null
161                }
162            ]
163        });
164        let resp: HeartbeatResponse = serde_json::from_value(json).unwrap();
165        assert_eq!(resp.pending_commands.len(), 1);
166        assert_eq!(resp.pending_commands[0].command, "trigger_sync");
167    }
168
169    #[test]
170    fn test_sync_result_default() {
171        let r = SyncResult::default();
172        assert_eq!(r.readings_synced, 0);
173        assert!(!r.full_sync);
174        assert!(r.errors.is_empty());
175    }
176
177    #[test]
178    fn test_sync_result_serialization_skips_empty() {
179        let r = SyncResult {
180            readings_synced: 100,
181            ..Default::default()
182        };
183        let json = serde_json::to_value(&r).unwrap();
184        assert_eq!(json["readings_synced"], 100);
185        assert!(json.get("errors").is_none());
186    }
187
188    #[test]
189    fn test_sync_event_create_serialization() {
190        let ev = SyncEventCreate {
191            service_id: Uuid::nil(),
192            command_id: None,
193            event_type: SyncEventType::Scheduled,
194            status: SyncEventStatus::Running,
195        };
196        let json = serde_json::to_value(&ev).unwrap();
197        assert_eq!(json["event_type"], "scheduled");
198        assert_eq!(json["status"], "running");
199        assert!(json.get("command_id").is_none());
200    }
201
202    #[test]
203    fn test_sync_event_update_skips_empty() {
204        let upd = SyncEventUpdate {
205            status: Some(SyncEventStatus::Completed),
206            readings_synced: Some(5),
207            ..Default::default()
208        };
209        let json = serde_json::to_value(&upd).unwrap();
210        assert_eq!(json["status"], "completed");
211        assert_eq!(json["readings_synced"], 5);
212        assert!(json.get("errors").is_none());
213        assert!(json.get("duration_ms").is_none());
214    }
215}
river_data_core/models/protocol.rs

river_data_core/models/
protocol.rs
