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

river_data_core/models/
protocol.rs
