agent_migrate_core/
watch.rs

1use std::path::Path;
2
3use serde::{Deserialize, Serialize};
4use sha2::{Digest, Sha256};
5use thiserror::Error;
6use uuid::Uuid;
7
8use crate::Endpoint;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum WatchDecision {
12    Apply,
13    Unchanged,
14    SuppressedOrigin,
15}
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct OriginStamp {
19    origin_id: Uuid,
20    content_hash: String,
21}
22
23impl OriginStamp {
24    #[must_use]
25    pub fn new(origin_id: Uuid, content_hash: impl Into<String>) -> Self {
26        Self {
27            origin_id,
28            content_hash: content_hash.into(),
29        }
30    }
31
32    #[must_use]
33    pub const fn origin_id(&self) -> Uuid {
34        self.origin_id
35    }
36
37    #[must_use]
38    pub fn matches(&self, content_hash: &str) -> bool {
39        self.content_hash == content_hash
40    }
41
42    /// Saves an origin stamp through an atomic same-directory replacement.
43    ///
44    /// # Errors
45    ///
46    /// Returns an error when serialization or durable filesystem operations fail.
47    pub fn save(&self, path: &Path) -> Result<(), WatchError> {
48        save_json(self, path)
49    }
50
51    /// Loads an origin stamp from JSON.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error when reading or parsing state fails.
56    pub fn load(path: &Path) -> Result<Self, WatchError> {
57        load_json(path)
58    }
59}
60
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub struct WatchState {
63    origin_id: Uuid,
64    source: Endpoint,
65    target: Endpoint,
66    last_applied_hash: Option<String>,
67}
68
69impl WatchState {
70    #[must_use]
71    pub fn new(source: Endpoint, target: Endpoint) -> Self {
72        let origin_id = relationship_id(&source, &target);
73        Self {
74            origin_id,
75            source,
76            target,
77            last_applied_hash: None,
78        }
79    }
80
81    #[must_use]
82    pub const fn origin_id(&self) -> Uuid {
83        self.origin_id
84    }
85
86    #[must_use]
87    pub fn source(&self) -> &Endpoint {
88        &self.source
89    }
90
91    #[must_use]
92    pub fn target(&self) -> &Endpoint {
93        &self.target
94    }
95
96    #[must_use]
97    pub fn decide(&self, hash: &str, incoming_origin: Option<Uuid>) -> WatchDecision {
98        if incoming_origin == Some(self.origin_id) {
99            WatchDecision::SuppressedOrigin
100        } else if self.last_applied_hash.as_deref() == Some(hash) {
101            WatchDecision::Unchanged
102        } else {
103            WatchDecision::Apply
104        }
105    }
106
107    pub fn record_applied(&mut self, hash: impl Into<String>) {
108        self.last_applied_hash = Some(hash.into());
109    }
110
111    /// Saves watcher state through an atomic same-directory replacement.
112    ///
113    /// # Errors
114    ///
115    /// Returns an error when serialization or durable filesystem operations fail.
116    pub fn save(&self, path: &Path) -> Result<(), WatchError> {
117        save_json(self, path)
118    }
119
120    /// Loads watcher state from its JSON file.
121    ///
122    /// # Errors
123    ///
124    /// Returns an error when reading or parsing state fails.
125    pub fn load(path: &Path) -> Result<Self, WatchError> {
126        load_json(path)
127    }
128}
129
130fn relationship_id(source: &Endpoint, target: &Endpoint) -> Uuid {
131    let source = source.to_string();
132    let target = target.to_string();
133    let (first, second) = if source <= target {
134        (source, target)
135    } else {
136        (target, source)
137    };
138    let digest = Sha256::digest(format!("{first}\0{second}").as_bytes());
139    let mut bytes = [0_u8; 16];
140    bytes.copy_from_slice(&digest[..16]);
141    Uuid::from_bytes(bytes)
142}
143
144#[must_use]
145pub fn content_hash(content: &[u8]) -> String {
146    hex::encode(Sha256::digest(content))
147}
148
149fn save_json(value: &impl Serialize, path: &Path) -> Result<(), WatchError> {
150    let parent = path.parent().ok_or(WatchError::MissingParent)?;
151    std::fs::create_dir_all(parent)?;
152    let mut temporary = tempfile::NamedTempFile::new_in(parent)?;
153    serde_json::to_writer_pretty(temporary.as_file_mut(), value)?;
154    temporary.as_file().sync_all()?;
155    temporary.persist(path).map_err(|error| error.error)?;
156    Ok(())
157}
158
159fn load_json<T: serde::de::DeserializeOwned>(path: &Path) -> Result<T, WatchError> {
160    let bytes = std::fs::read(path)?;
161    serde_json::from_slice(&bytes).map_err(Into::into)
162}
163
164#[derive(Debug, Error)]
165pub enum WatchError {
166    #[error("watch state I/O failed")]
167    Io(#[from] std::io::Error),
168    #[error("watch state JSON is invalid")]
169    Json(#[from] serde_json::Error),
170    #[error("watch state path has no parent directory")]
171    MissingParent,
172}
agent_migrate_core/watch.rs

agent_migrate_core/
watch.rs
