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

river_data_core/models/
protocol.rs
