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

river_data_core/models/
protocol.rs
