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    /// Readings withheld pending audit acknowledgement; the API re-sends them
112    /// on the next incremental fetch.
113    #[serde(default)]
114    pub readings_held: u64,
115    pub status_events_synced: u64,
116    pub full_sync: bool,
117    pub duration_ms: u64,
118    #[serde(skip_serializing_if = "Vec::is_empty")]
119    pub errors: Vec<String>,
120    #[serde(skip_serializing_if = "Vec::is_empty")]
121    pub log: Vec<String>,
122}
123
124#[derive(Debug)]
125pub enum SyncTrigger {
126    Scheduled,
127    Command { id: Uuid, full: bool },
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133
134    #[test]
135    fn test_enroll_request_serialization() {
136        let req = EnrollRequest {
137            client_id: "svc_abc".to_string(),
138            client_secret: "secret123".to_string(),
139            instance_id: "service-01".to_string(),
140        };
141        let json = serde_json::to_value(&req).unwrap();
142        assert_eq!(json["client_id"], "svc_abc");
143        assert_eq!(json["instance_id"], "service-01");
144    }
145
146    #[test]
147    fn test_enroll_response_deserialization() {
148        let json = serde_json::json!({
149            "service_id": "550e8400-e29b-41d4-a716-446655440000",
150            "session_token": "tok-abc"
151        });
152        let resp: EnrollResponse = serde_json::from_value(json).unwrap();
153        assert_eq!(resp.session_token, "tok-abc");
154    }
155
156    #[test]
157    fn test_heartbeat_response_with_commands() {
158        let json = serde_json::json!({
159            "session_token": "new-tok",
160            "pending_commands": [
161                {
162                    "id": "550e8400-e29b-41d4-a716-446655440000",
163                    "command": "trigger_sync",
164                    "payload": null
165                }
166            ]
167        });
168        let resp: HeartbeatResponse = serde_json::from_value(json).unwrap();
169        assert_eq!(resp.pending_commands.len(), 1);
170        assert_eq!(resp.pending_commands[0].command, "trigger_sync");
171    }
172
173    #[test]
174    fn test_sync_result_default() {
175        let r = SyncResult::default();
176        assert_eq!(r.readings_synced, 0);
177        assert!(!r.full_sync);
178        assert!(r.errors.is_empty());
179    }
180
181    #[test]
182    fn test_sync_result_serialization_skips_empty() {
183        let r = SyncResult {
184            readings_synced: 100,
185            ..Default::default()
186        };
187        let json = serde_json::to_value(&r).unwrap();
188        assert_eq!(json["readings_synced"], 100);
189        assert!(json.get("errors").is_none());
190    }
191
192    #[test]
193    fn test_sync_event_create_serialization() {
194        let ev = SyncEventCreate {
195            service_id: Uuid::nil(),
196            command_id: None,
197            event_type: SyncEventType::Scheduled,
198            status: SyncEventStatus::Running,
199        };
200        let json = serde_json::to_value(&ev).unwrap();
201        assert_eq!(json["event_type"], "scheduled");
202        assert_eq!(json["status"], "running");
203        assert!(json.get("command_id").is_none());
204    }
205
206    #[test]
207    fn test_sync_event_update_skips_empty() {
208        let upd = SyncEventUpdate {
209            status: Some(SyncEventStatus::Completed),
210            readings_synced: Some(5),
211            ..Default::default()
212        };
213        let json = serde_json::to_value(&upd).unwrap();
214        assert_eq!(json["status"], "completed");
215        assert_eq!(json["readings_synced"], 5);
216        assert!(json.get("errors").is_none());
217        assert!(json.get("duration_ms").is_none());
218    }
219}
river_data_core/models/protocol.rs

river_data_core/models/
protocol.rs
