qualia_client_core/identity_plane/
fleet_jobs.rs

1//! Multi-apparatus job delivery: sign, POST, accept, outbox.
2//!
3//! Local jobs stay on the local queue. Jobs aimed at a registered remote
4//! apparatus are signed by the person principal and delivered to that device's
5//! `control_base_url` (`POST /api/fleet/jobs`). Failures land in a durable
6//! outbox for retry. Unknown targets still fail closed.
7
8use super::fleet::{ensure_local_apparatus, resolve_job_placement, JobPlacement};
9use super::person::PersonPrincipal;
10use crate::local_job_scheduler::{LocalJob, LocalJobKind, LocalJobScheduler};
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13use std::path::PathBuf;
14use std::time::{SystemTime, UNIX_EPOCH};
15
16pub const FLEET_JOB_FORMAT: &str = "qualia.fleet.job.v1";
17
18#[derive(Debug, Clone, Serialize, Deserialize)]
19pub struct FleetJobEnvelope {
20    pub format: String,
21    pub person_id: String,
22    pub person_verifying_key_hex: String,
23    pub source_device_id: String,
24    pub target_device_id: String,
25    pub kind: LocalJobKind,
26    pub created_at_unix: u64,
27    /// Hex-encoded Ed25519 signature over [`signing_payload`].
28    pub signature_hex: String,
29}
30
31#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct RemoteOutboxEntry {
33    pub id: String,
34    pub envelope: FleetJobEnvelope,
35    pub target_url: String,
36    pub attempts: u32,
37    pub last_error: Option<String>,
38    pub last_attempt_unix: u64,
39    pub created_at_unix: u64,
40    pub delivered: bool,
41}
42
43fn now_unix() -> u64 {
44    SystemTime::now()
45        .duration_since(UNIX_EPOCH)
46        .map(|d| d.as_secs())
47        .unwrap_or(0)
48}
49
50fn outbox_path() -> PathBuf {
51    crate::state::app_meta_dir().join("remote_job_outbox.json")
52}
53
54fn load_outbox() -> Vec<RemoteOutboxEntry> {
55    let path = outbox_path();
56    std::fs::read_to_string(path)
57        .ok()
58        .and_then(|s| serde_json::from_str(&s).ok())
59        .unwrap_or_default()
60}
61
62fn save_outbox(entries: &[RemoteOutboxEntry]) -> Result<(), String> {
63    let path = outbox_path();
64    if let Some(parent) = path.parent() {
65        std::fs::create_dir_all(parent).map_err(|e| e.to_string())?;
66    }
67    let json = serde_json::to_string_pretty(entries).map_err(|e| e.to_string())?;
68    std::fs::write(path, json).map_err(|e| e.to_string())
69}
70
71/// Canonical bytes signed by the person principal.
72pub fn signing_payload(
73    person_id: &str,
74    source_device_id: &str,
75    target_device_id: &str,
76    kind: &LocalJobKind,
77    created_at_unix: u64,
78) -> Result<Vec<u8>, String> {
79    let kind_json = serde_json::to_string(kind).map_err(|e| e.to_string())?;
80    let mut h = Sha256::new();
81    h.update(FLEET_JOB_FORMAT.as_bytes());
82    h.update(b"|");
83    h.update(person_id.as_bytes());
84    h.update(b"|");
85    h.update(source_device_id.as_bytes());
86    h.update(b"|");
87    h.update(target_device_id.as_bytes());
88    h.update(b"|");
89    h.update(created_at_unix.to_le_bytes());
90    h.update(b"|");
91    h.update(kind_json.as_bytes());
92    Ok(h.finalize().to_vec())
93}
94
95impl FleetJobEnvelope {
96    pub fn build(
97        person: &PersonPrincipal,
98        source_device_id: &str,
99        target_device_id: &str,
100        kind: LocalJobKind,
101    ) -> Result<Self, String> {
102        let created_at_unix = now_unix();
103        let payload = signing_payload(
104            &person.person_id,
105            source_device_id,
106            target_device_id,
107            &kind,
108            created_at_unix,
109        )?;
110        let sig = person.sign_message(&payload);
111        Ok(Self {
112            format: FLEET_JOB_FORMAT.to_string(),
113            person_id: person.person_id.clone(),
114            person_verifying_key_hex: person.verifying_key_hex(),
115            source_device_id: source_device_id.to_string(),
116            target_device_id: target_device_id.to_string(),
117            kind,
118            created_at_unix,
119            signature_hex: hex::encode(sig),
120        })
121    }
122
123    pub fn verify(&self) -> Result<(), String> {
124        if self.format != FLEET_JOB_FORMAT {
125            return Err(format!("unsupported fleet job format: {}", self.format));
126        }
127        let payload = signing_payload(
128            &self.person_id,
129            &self.source_device_id,
130            &self.target_device_id,
131            &self.kind,
132            self.created_at_unix,
133        )?;
134        let sig_bytes = hex::decode(&self.signature_hex).map_err(|e| e.to_string())?;
135        if sig_bytes.len() != 64 {
136            return Err("signature must be 64 bytes".into());
137        }
138        let mut sig = [0u8; 64];
139        sig.copy_from_slice(&sig_bytes);
140        PersonPrincipal::verify_message(
141            &self.person_id,
142            &self.person_verifying_key_hex,
143            &payload,
144            &sig,
145        )
146    }
147}
148
149/// Deliver a job to a remote apparatus, or queue to outbox on failure.
150pub fn deliver_or_queue_remote_job(
151    kind: LocalJobKind,
152    target_device_id: &str,
153) -> Result<RemoteOutboxEntry, String> {
154    let plane = ensure_local_apparatus(None)?;
155    let placement = resolve_job_placement(Some(target_device_id))?;
156    let (device_id, label) = match placement {
157        JobPlacement::RemoteRegistered { device_id, label } => (device_id, label),
158        JobPlacement::Local { .. } => {
159            return Err("target is the local apparatus — use the local job queue".into());
160        }
161        JobPlacement::Unknown { device_id } => {
162            return Err(format!("unknown device {device_id}"));
163        }
164    };
165    let peer = plane
166        .devices
167        .iter()
168        .find(|d| d.device_id == device_id)
169        .ok_or("peer missing from fleet")?;
170    if peer.control_base_url.trim().is_empty() {
171        return Err(format!(
172            "Peer '{label}' has no control_base_url. On that machine, set a LAN URL (Settings → Person & devices) so jobs can be delivered."
173        ));
174    }
175    let person = PersonPrincipal::load_or_create(None)?;
176    let envelope = FleetJobEnvelope::build(&person, &plane.local_device_id, &device_id, kind)?;
177    let url = format!(
178        "{}/api/fleet/jobs",
179        peer.control_base_url.trim_end_matches('/')
180    );
181    let mut entry = RemoteOutboxEntry {
182        id: uuid::Uuid::new_v4().to_string(),
183        envelope: envelope.clone(),
184        target_url: url.clone(),
185        attempts: 0,
186        last_error: None,
187        last_attempt_unix: 0,
188        created_at_unix: now_unix(),
189        delivered: false,
190    };
191
192    match try_deliver(&url, &envelope) {
193        Ok(()) => {
194            entry.delivered = true;
195            entry.attempts = 1;
196            entry.last_attempt_unix = now_unix();
197        }
198        Err(e) => {
199            entry.attempts = 1;
200            entry.last_error = Some(e);
201            entry.last_attempt_unix = now_unix();
202            let mut box_ = load_outbox();
203            box_.push(entry.clone());
204            // Cap outbox
205            if box_.len() > 64 {
206                box_.retain(|e| !e.delivered);
207                if box_.len() > 64 {
208                    box_.drain(0..box_.len() - 64);
209                }
210            }
211            save_outbox(&box_)?;
212        }
213    }
214    if entry.delivered {
215        // Keep a short delivered audit trail
216        let mut box_ = load_outbox();
217        box_.push(entry.clone());
218        if box_.len() > 64 {
219            box_.drain(0..box_.len() - 64);
220        }
221        let _ = save_outbox(&box_);
222    }
223    Ok(entry)
224}
225
226fn try_deliver(url: &str, envelope: &FleetJobEnvelope) -> Result<(), String> {
227    let client = reqwest::blocking::Client::builder()
228        .timeout(std::time::Duration::from_secs(15))
229        .build()
230        .map_err(|e| e.to_string())?;
231    let resp = client
232        .post(url)
233        .json(envelope)
234        .send()
235        .map_err(|e| format!("fleet deliver: {e}"))?;
236    if !resp.status().is_success() {
237        let status = resp.status();
238        let body = resp.text().unwrap_or_default();
239        return Err(format!("fleet deliver HTTP {status}: {body}"));
240    }
241    Ok(())
242}
243
244/// Accept a signed fleet job on this apparatus (HTTP handler body).
245pub fn accept_fleet_job_envelope(envelope: FleetJobEnvelope) -> Result<LocalJob, String> {
246    envelope.verify()?;
247    let plane = ensure_local_apparatus(None)?;
248    if envelope.target_device_id != plane.local_device_id {
249        return Err(format!(
250            "job targets {} but this apparatus is {}",
251            envelope.target_device_id, plane.local_device_id
252        ));
253    }
254    // Same person principal required (imported on both machines).
255    if envelope.person_id != plane.person.person_id {
256        return Err(
257            "fleet job person_id does not match this install's person principal — import the same person transfer bundle on both machines"
258                .into(),
259        );
260    }
261    // Enqueue as local work; placement already verified as this apparatus.
262    let mut job = LocalJobScheduler::global()
263        .enqueue_for_device(envelope.kind, Some(plane.local_device_id.clone()))?;
264    job.originating_device_id = Some(envelope.source_device_id.clone());
265    job.person_id = Some(envelope.person_id.clone());
266    job.message = format!("Accepted from fleet peer {}", envelope.source_device_id);
267    LocalJobScheduler::global().update_job_meta(&job)?;
268    Ok(job)
269}
270
271pub fn list_remote_outbox() -> Result<Vec<RemoteOutboxEntry>, String> {
272    Ok(load_outbox())
273}
274
275/// Retry undelivered outbox entries (best-effort).
276pub fn retry_remote_outbox() -> Result<usize, String> {
277    let mut box_ = load_outbox();
278    let mut delivered = 0usize;
279    for entry in box_.iter_mut() {
280        if entry.delivered {
281            continue;
282        }
283        entry.attempts = entry.attempts.saturating_add(1);
284        entry.last_attempt_unix = now_unix();
285        match try_deliver(&entry.target_url, &entry.envelope) {
286            Ok(()) => {
287                entry.delivered = true;
288                entry.last_error = None;
289                delivered += 1;
290            }
291            Err(e) => entry.last_error = Some(e),
292        }
293    }
294    save_outbox(&box_)?;
295    Ok(delivered)
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301
302    #[test]
303    fn envelope_round_trip_verifies() {
304        let person = PersonPrincipal::generate("t").unwrap();
305        let kind = LocalJobKind::DaemonGraphReload;
306        let env = FleetJobEnvelope::build(&person, "did:q42:device:aa", "did:q42:device:bb", kind)
307            .unwrap();
308        env.verify().unwrap();
309        let mut bad = env.clone();
310        bad.signature_hex = "00".repeat(64);
311        assert!(bad.verify().is_err());
312    }
313}
qualia_client_core/identity_plane/fleet_jobs.rs

qualia_client_core/identity_plane/
fleet_jobs.rs
