vissue_tui/
control.rs

1//! Socket-backed board. Unix only; clients never bind the control socket.
2
3use std::path::Path;
4use std::sync::Mutex;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::time::Duration;
7
8use serde_json::Value;
9use vissue_control::client::Client;
10use vissue_control::rpc::{
11    CONFLICT, CYCLE, ClaimParams, Error as RpcError, INVALID_STATE, IdParams, InitializeResult,
12    IssueListParams, IssueListResult, MutResult as WireMut, NOT_FOUND, NoteParams, Notification,
13    RelatedParams, Request, SearchParams, TreeParams, UpdateParams,
14};
15use vissue_control::{InitializeParams, PROTOCOL_VERSION};
16use vissue_core::config::Layout;
17use vissue_core::error::Error;
18use vissue_core::views::{
19    AgendaRow, ClaimRow, Excerpt, IssueDetail, ListQuery, RelatedHit, SearchHit, TreeNode,
20};
21
22use crate::backend::{BackendKind, BoardBackend, ListPage, MutResult, SinceGate, UpdateReq};
23
24/// JSON-RPC client after a matching `initialize`.
25#[derive(Debug)]
26pub struct ControlBackend {
27    layout: Layout,
28    identity: String,
29    client: Mutex<Client>,
30    generation: AtomicU64,
31    revision: AtomicU64,
32    /// Revision of the last full list/ready page, not the last mut/notify.
33    page_revision: AtomicU64,
34    since: SinceGate,
35    last_query: Mutex<Option<ListQuery>>,
36    last_since: Mutex<Option<Option<u64>>>,
37}
38
39impl ControlBackend {
40    /// Connect, `initialize` with a required agent, and refuse a root/prefix
41    /// mismatch so mutations never hit the wrong vault.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if the socket cannot be reached, `initialize` fails, or
46    /// the serve root/prefix does not match `layout`.
47    pub fn connect(path: &Path, layout: &Layout, agent: &str) -> Result<Self, ControlAttachError> {
48        Self::connect_as(path, layout, agent, "vissue-tui")
49    }
50
51    /// Same as [`Self::connect`] with an explicit `initialize.client` name.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error if the socket cannot be reached, `initialize` fails, or
56    /// the serve root/prefix does not match `layout`.
57    pub fn connect_as(
58        path: &Path,
59        layout: &Layout,
60        agent: &str,
61        client: &str,
62    ) -> Result<Self, ControlAttachError> {
63        let mut client_conn = Client::connect(path).map_err(ControlAttachError::Rpc)?;
64        let params = InitializeParams {
65            protocol_version: PROTOCOL_VERSION,
66            client: client.into(),
67            agent: agent.to_string(),
68        };
69        let value = client_conn
70            .request_typed(&Request::Initialize(params))
71            .map_err(ControlAttachError::Rpc)?;
72        let init: InitializeResult = serde_json::from_value(value)
73            .map_err(|e| ControlAttachError::Rpc(RpcError::Json(e)))?;
74        if !roots_match(layout, &init.root, &init.prefix) {
75            return Err(ControlAttachError::Mismatch {
76                want_root: layout.root().display().to_string(),
77                want_prefix: layout.prefix().to_string(),
78                got_root: init.root,
79                got_prefix: init.prefix,
80            });
81        }
82        Ok(Self {
83            layout: layout.clone(),
84            identity: init.identity,
85            client: Mutex::new(client_conn),
86            generation: AtomicU64::new(init.generation),
87            revision: AtomicU64::new(init.revision),
88            page_revision: AtomicU64::new(0),
89            since: SinceGate::after_attach(),
90            last_query: Mutex::new(None),
91            last_since: Mutex::new(None),
92        })
93    }
94
95    fn call(&self, req: &Request) -> Result<Value, Error> {
96        let mut client = self.client.lock().expect("control client");
97        client.request_typed(req).map_err(map_rpc)
98    }
99
100    fn list_params(&self, q: ListQuery) -> IssueListParams {
101        // Serve `unchanged` is catalog-wide. Only send since_revision when
102        // this is the same ready/project/query as the last full page.
103        let mut last_query = self.last_query.lock().expect("query");
104        let same = last_query.as_ref() == Some(&q);
105        *last_query = Some(q.clone());
106        drop(last_query);
107        let page = self.page_revision.load(Ordering::SeqCst);
108        let since = if same {
109            self.since.next(page)
110        } else {
111            self.since.invalidate();
112            let _ = self.since.next(page);
113            None
114        };
115        *self.last_since.lock().expect("since") = Some(since);
116        IssueListParams {
117            project: q.project,
118            state: q.state,
119            ready: if q.ready { Some(true) } else { None },
120            query: q.query,
121            limit: q.limit,
122            offset: q.offset,
123            since_revision: since,
124        }
125    }
126
127    fn apply_list(&self, result: IssueListResult) -> ListPage {
128        if !result.unchanged {
129            self.page_revision.store(result.revision, Ordering::SeqCst);
130            self.revision.store(result.revision, Ordering::SeqCst);
131            self.generation.store(result.generation, Ordering::SeqCst);
132        }
133        ListPage {
134            issues: result.issues,
135            total: result.total,
136            matched: result.matched,
137            revision: result.revision,
138            generation: result.generation,
139            unchanged: result.unchanged,
140        }
141    }
142
143    fn apply_mut(&self, wire: WireMut) -> MutResult {
144        self.revision.store(wire.revision, Ordering::SeqCst);
145        self.generation.store(wire.generation, Ordering::SeqCst);
146        MutResult {
147            ok: wire.ok,
148            report: wire.report,
149            issue: wire.issue,
150            revision: wire.revision,
151            generation: wire.generation,
152        }
153    }
154}
155
156fn roots_match(layout: &Layout, root: &str, prefix: &str) -> bool {
157    let want_root = layout.root().display().to_string();
158    (root == want_root || Path::new(root) == layout.root()) && prefix == layout.prefix()
159}
160
161/// Why attach refused the live socket.
162#[derive(Debug)]
163pub enum ControlAttachError {
164    /// Socket, framing, or JSON-RPC failure.
165    Rpc(RpcError),
166    /// Serve answered for a different vault than `layout`.
167    Mismatch {
168        /// Board layout root.
169        want_root: String,
170        /// Board layout prefix.
171        want_prefix: String,
172        /// Root `initialize` returned.
173        got_root: String,
174        /// Prefix `initialize` returned.
175        got_prefix: String,
176    },
177}
178
179impl std::fmt::Display for ControlAttachError {
180    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
181        match self {
182            Self::Rpc(err) => write!(f, "{err}"),
183            Self::Mismatch {
184                want_root,
185                want_prefix,
186                got_root,
187                got_prefix,
188            } => write!(
189                f,
190                "serve root/prefix mismatch: want {want_root} {want_prefix}, got {got_root} {got_prefix}"
191            ),
192        }
193    }
194}
195
196impl std::error::Error for ControlAttachError {}
197
198fn map_rpc(err: RpcError) -> Error {
199    match err {
200        RpcError::Rpc(rpc) => match rpc.code {
201            NOT_FOUND => Error::IssueNotFound {
202                id: rpc
203                    .data
204                    .as_ref()
205                    .and_then(|d| d.get("id"))
206                    .and_then(Value::as_str)
207                    .unwrap_or("")
208                    .to_string(),
209            },
210            CONFLICT
211                if rpc
212                    .data
213                    .as_ref()
214                    .and_then(|d| d.get("code"))
215                    .and_then(Value::as_str)
216                    == Some("duplicate_id") =>
217            {
218                Error::DuplicateId {
219                    id: rpc
220                        .data
221                        .as_ref()
222                        .and_then(|d| d.get("id"))
223                        .and_then(Value::as_str)
224                        .unwrap_or("")
225                        .to_string(),
226                    paths: rpc
227                        .data
228                        .as_ref()
229                        .and_then(|d| d.get("paths"))
230                        .and_then(Value::as_array)
231                        .map(|arr| {
232                            arr.iter()
233                                .filter_map(Value::as_str)
234                                .map(std::path::PathBuf::from)
235                                .collect()
236                        })
237                        .unwrap_or_default(),
238                }
239            }
240            CONFLICT => Error::ClaimConflict {
241                id: rpc
242                    .data
243                    .as_ref()
244                    .and_then(|d| d.get("id"))
245                    .and_then(Value::as_str)
246                    .unwrap_or("")
247                    .to_string(),
248                holder: rpc
249                    .data
250                    .as_ref()
251                    .and_then(|d| d.get("holder"))
252                    .and_then(Value::as_str)
253                    .unwrap_or("")
254                    .to_string(),
255                claimed_at: None,
256            },
257            CYCLE => Error::BlockerCycle {
258                blocker: rpc
259                    .data
260                    .as_ref()
261                    .and_then(|d| d.get("block"))
262                    .and_then(Value::as_str)
263                    .unwrap_or("")
264                    .to_string(),
265                issue: rpc
266                    .data
267                    .as_ref()
268                    .and_then(|d| d.get("id"))
269                    .and_then(Value::as_str)
270                    .unwrap_or("")
271                    .to_string(),
272            },
273            INVALID_STATE => Error::InvalidState {
274                id: rpc
275                    .data
276                    .as_ref()
277                    .and_then(|d| d.get("id"))
278                    .and_then(Value::as_str)
279                    .unwrap_or("")
280                    .to_string(),
281                state: rpc
282                    .data
283                    .as_ref()
284                    .and_then(|d| d.get("state"))
285                    .and_then(Value::as_str)
286                    .unwrap_or("")
287                    .to_string(),
288            },
289            _ => Error::Other(anyhow::anyhow!("{}", rpc.message)),
290        },
291        other => Error::Other(anyhow::anyhow!("{other}")),
292    }
293}
294
295fn decode<T: serde::de::DeserializeOwned>(value: Value) -> Result<T, Error> {
296    serde_json::from_value(value).map_err(|e| Error::Other(e.into()))
297}
298
299impl BoardBackend for ControlBackend {
300    fn layout(&self) -> &Layout {
301        &self.layout
302    }
303
304    fn generation(&self) -> u64 {
305        self.generation.load(Ordering::SeqCst)
306    }
307
308    fn revision(&self) -> u64 {
309        self.revision.load(Ordering::SeqCst)
310    }
311
312    fn live(&self) -> BackendKind {
313        BackendKind::Control
314    }
315
316    fn identity(&self) -> &str {
317        &self.identity
318    }
319
320    fn list(&self, q: ListQuery) -> Result<ListPage, Error> {
321        let params = self.list_params(q);
322        let value = self.call(&Request::IssueList(params))?;
323        Ok(self.apply_list(decode(value)?))
324    }
325
326    fn ready(&self, project: Option<&str>) -> Result<ListPage, Error> {
327        let params = self.list_params(ListQuery {
328            project: project.map(str::to_string),
329            ready: true,
330            ..ListQuery::default()
331        });
332        let value = self.call(&Request::IssueReady(params))?;
333        Ok(self.apply_list(decode(value)?))
334    }
335
336    fn get(&self, id: &str) -> Result<IssueDetail, Error> {
337        let value = self.call(&Request::IssueGet(IdParams { id: id.to_string() }))?;
338        let row: vissue_control::rpc::IssueGetResult = decode(value)?;
339        Ok(row.issue)
340    }
341
342    fn excerpt(&self, id: &str) -> Result<Excerpt, Error> {
343        let value = self.call(&Request::IssueExcerpt(IdParams { id: id.to_string() }))?;
344        decode(value)
345    }
346
347    fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchHit>, Error> {
348        let value = self.call(&Request::IssueSearch(SearchParams {
349            query: query.to_string(),
350            limit: Some(limit),
351        }))?;
352        decode(value)
353    }
354
355    fn claims(&self, holder: Option<&str>, project: Option<&str>) -> Result<Vec<ClaimRow>, Error> {
356        let value = self.call(&Request::IssueClaims(vissue_control::rpc::ClaimsParams {
357            holder: holder.map(str::to_string),
358            project: project.map(str::to_string),
359        }))?;
360        decode(value)
361    }
362
363    fn agenda(&self, days: i64, project: Option<&str>) -> Result<Vec<AgendaRow>, Error> {
364        let value = self.call(&Request::IssueAgenda(vissue_control::rpc::AgendaParams {
365            days: Some(days),
366            project: project.map(str::to_string),
367        }))?;
368        decode(value)
369    }
370
371    fn tree(&self, id: &str) -> Result<TreeNode, Error> {
372        let value = self.call(&Request::IssueTree(TreeParams {
373            id: id.to_string(),
374            format: Some("nodes".into()),
375        }))?;
376        match decode::<vissue_control::rpc::TreeResult>(value)? {
377            vissue_control::rpc::TreeResult::Nodes(node) => Ok(node),
378            vissue_control::rpc::TreeResult::Text { text } => Err(Error::Other(anyhow::anyhow!(
379                "serve returned tree text, not nodes: {text}"
380            ))),
381        }
382    }
383
384    fn related(&self, id: &str, depth: usize, limit: usize) -> Result<Vec<RelatedHit>, Error> {
385        let value = self.call(&Request::IssueRelated(RelatedParams {
386            id: id.to_string(),
387            depth: Some(depth),
388            limit: Some(limit),
389        }))?;
390        decode(value)
391    }
392
393    fn projects(&self) -> Result<Vec<String>, Error> {
394        let value = self.call(&Request::ProjectList)?;
395        let row: vissue_control::rpc::ProjectListResult = decode(value)?;
396        Ok(row.projects)
397    }
398
399    fn claim(&self, id: &str, force: bool) -> Result<MutResult, Error> {
400        let value = self.call(&Request::IssueClaim(ClaimParams {
401            id: id.to_string(),
402            force,
403            agent: None,
404        }))?;
405        Ok(self.apply_mut(decode(value)?))
406    }
407
408    fn note(&self, id: &str, text: &str) -> Result<MutResult, Error> {
409        let value = self.call(&Request::IssueNote(NoteParams {
410            id: id.to_string(),
411            text: text.to_string(),
412        }))?;
413        Ok(self.apply_mut(decode(value)?))
414    }
415
416    fn update(&self, req: UpdateReq) -> Result<MutResult, Error> {
417        let value = self.call(&Request::IssueUpdate(UpdateParams {
418            id: req.id,
419            state: req.state,
420            priority: req.priority.map(|c| c.to_string()),
421            block: req.block,
422            unblock: req.unblock,
423            if_state: req.if_state,
424            if_gen: req.if_gen,
425            agent: None,
426        }))?;
427        Ok(self.apply_mut(decode(value)?))
428    }
429
430    fn open(&self, id: &str) -> Result<IssueDetail, Error> {
431        let value = self.call(&Request::IssueOpen(IdParams { id: id.to_string() }))?;
432        let row: vissue_control::rpc::IssueGetResult = decode(value)?;
433        Ok(row.issue)
434    }
435
436    /// # Panics
437    ///
438    /// Panics if the control client lock is poisoned.
439    fn wait(&self, last: u64, timeout_ms: u64) -> Result<u64, Error> {
440        let mut client = self.client.lock().expect("control client");
441        match client.wait_notification(Duration::from_millis(timeout_ms.max(1))) {
442            Ok(Notification::VaultChanged(changed)) => {
443                self.revision.store(changed.revision, Ordering::SeqCst);
444                self.generation.store(changed.generation, Ordering::SeqCst);
445                Ok(changed.revision)
446            }
447            Ok(_) => Ok(self.revision.load(Ordering::SeqCst)),
448            Err(_) => Ok(last),
449        }
450    }
451
452    /// # Panics
453    ///
454    /// Panics if the since lock is poisoned.
455    fn last_since_revision(&self) -> Option<Option<u64>> {
456        *self.last_since.lock().expect("since")
457    }
458
459    /// # Panics
460    ///
461    /// Panics if the query lock is poisoned.
462    fn invalidate_since(&self) {
463        self.since.invalidate();
464        *self.last_query.lock().expect("query") = None;
465    }
466}
467
468#[cfg(test)]
469mod tests {
470    use super::*;
471    use crate::backend::{BoardBackend, UpdateReq};
472    use serde_json::json;
473    use std::io::{BufReader, Write};
474    use std::os::unix::net::UnixListener;
475    use std::sync::{Arc, Mutex};
476    use std::thread;
477    use vissue_control::frame::{read_message, write_message};
478    use vissue_control::rpc::JsonRpcRequest;
479    use vissue_core::views::ListQuery;
480
481    #[test]
482    fn after_initialize_the_next_list_omits_since_revision() {
483        let dir = tempfile::tempdir().unwrap();
484        let sock = dir.path().join("control.sock");
485        let layout = Layout::new(dir.path().join("vault"), "Software");
486        let seen = Arc::new(Mutex::new(Vec::new()));
487        let seen_cb = Arc::clone(&seen);
488        let root = layout.root().display().to_string();
489        let listener = UnixListener::bind(&sock).unwrap();
490        thread::spawn(move || {
491            let (stream, _) = listener.accept().unwrap();
492            let mut reader = BufReader::new(stream.try_clone().unwrap());
493            let mut writer = stream;
494            while let Ok((payload, framing)) = read_message(&mut reader) {
495                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
496                let body = if req.method == "initialize" {
497                    json!({
498                        "jsonrpc": "2.0",
499                        "id": req.id,
500                        "result": {
501                            "protocolVersion": 1,
502                            "capabilities": [],
503                            "root": root,
504                            "prefix": "Software",
505                            "generation": 9,
506                            "revision": 41,
507                            "identity": "tui"
508                        }
509                    })
510                } else {
511                    let since = req
512                        .params
513                        .as_ref()
514                        .and_then(|p| p.get("since_revision"))
515                        .cloned();
516                    seen_cb.lock().unwrap().push(since);
517                    json!({
518                        "jsonrpc": "2.0",
519                        "id": req.id,
520                        "result": {
521                            "issues": [],
522                            "total": 0,
523                            "matched": 0,
524                            "revision": 41,
525                            "generation": 9,
526                            "unchanged": false
527                        }
528                    })
529                };
530                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
531                writer.flush().unwrap();
532            }
533        });
534
535        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
536        assert_eq!(backend.revision(), 41);
537        assert_eq!(backend.live(), BackendKind::Control);
538        backend.ready(None).unwrap();
539        assert_eq!(backend.last_since_revision(), Some(None));
540        backend.ready(None).unwrap();
541        assert_eq!(backend.last_since_revision(), Some(Some(41)));
542        backend.list(ListQuery::default()).unwrap();
543        assert_eq!(backend.last_since_revision(), Some(None));
544        let seen = seen.lock().unwrap();
545        assert_eq!(seen.len(), 3);
546        assert_eq!(seen[0], None);
547        assert_eq!(seen[1], Some(json!(41)));
548        assert_eq!(seen[2], None);
549    }
550
551    fn serve_methods(path: &std::path::Path, root: String) {
552        let listener = UnixListener::bind(path).unwrap();
553        thread::spawn(move || {
554            let (stream, _) = listener.accept().unwrap();
555            let mut reader = BufReader::new(stream.try_clone().unwrap());
556            let mut writer = stream;
557            while let Ok((payload, framing)) = read_message(&mut reader) {
558                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
559                let result = match req.method.as_str() {
560                    "initialize" => json!({
561                        "protocolVersion":1,"capabilities":[],"root":root,
562                        "prefix":"Software","generation":2,"revision":3,"identity":"tui"
563                    }),
564                    "issue/get" | "issue/show" | "issue/open" => json!({
565                        "id":"atlas-1a2b","project":"atlas","title":"t","state":"TODO",
566                        "priority":"B","properties":{},"org_tags":[],"tags":[],
567                        "blocked_by":[],"parent":null,"claimed_by":null,"claimed_at":null,
568                        "file":"f","line_start":1,"line_end":2,"revision":3
569                    }),
570                    "issue/excerpt" => json!({
571                        "id":"atlas-1a2b","file":"f","line_start":1,"line_end":2,
572                        "text":"body","suppressed":false
573                    }),
574                    "issue/search" | "issue/claims" | "issue/agenda" | "issue/related" => {
575                        json!([])
576                    }
577                    "issue/tree" => json!({
578                        "id":"atlas-1a2b","state":"TODO","title":"t",
579                        "children":[],"blocked_by":[]
580                    }),
581                    "project/list" => json!({"projects":["atlas"],"revision":3}),
582                    "issue/claim" | "issue/note" | "issue/update" => json!({
583                        "ok":true,"report":"ok","issue":null,"revision":4,"generation":3
584                    }),
585                    "issue/list" | "issue/ready" => json!({
586                        "issues":[],"total":0,"matched":0,"revision":3,
587                        "generation":2,"unchanged":false
588                    }),
589                    other => panic!("unexpected {other}"),
590                };
591                let body = json!({"jsonrpc":"2.0","id":req.id,"result":result});
592                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
593                writer.flush().unwrap();
594            }
595        });
596    }
597
598    #[test]
599    fn control_verbs_roundtrip() {
600        let dir = tempfile::tempdir().unwrap();
601        let sock = dir.path().join("control.sock");
602        let layout = Layout::new(dir.path().join("vault"), "Software");
603        serve_methods(&sock, layout.root().display().to_string());
604        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
605        assert_eq!(backend.get("atlas-1a2b").unwrap().id, "atlas-1a2b");
606        assert_eq!(backend.excerpt("atlas-1a2b").unwrap().text, "body");
607        assert!(backend.search("x", 5).unwrap().is_empty());
608        assert!(backend.claims(None, None).unwrap().is_empty());
609        assert!(backend.agenda(14, None).unwrap().is_empty());
610        assert_eq!(backend.tree("atlas-1a2b").unwrap().id, "atlas-1a2b");
611        assert!(backend.related("atlas-1a2b", 2, 5).unwrap().is_empty());
612        assert_eq!(backend.projects().unwrap(), ["atlas"]);
613        assert!(backend.claim("atlas-1a2b", false).unwrap().ok);
614        assert!(backend.note("atlas-1a2b", "hi").unwrap().ok);
615        assert!(
616            backend
617                .update(UpdateReq {
618                    id: "atlas-1a2b".into(),
619                    state: Some("STARTED".into()),
620                    ..UpdateReq::default()
621                })
622                .unwrap()
623                .ok
624        );
625        assert_eq!(backend.open("atlas-1a2b").unwrap().id, "atlas-1a2b");
626        assert_eq!(backend.wait(3, 5).unwrap(), 3);
627    }
628
629    #[test]
630    fn after_claim_next_list_sends_page_revision_not_head() {
631        let dir = tempfile::tempdir().unwrap();
632        let sock = dir.path().join("control.sock");
633        let layout = Layout::new(dir.path().join("vault"), "Software");
634        let seen = Arc::new(Mutex::new(Vec::new()));
635        let seen_cb = Arc::clone(&seen);
636        let root = layout.root().display().to_string();
637        let listener = UnixListener::bind(&sock).unwrap();
638        thread::spawn(move || {
639            let (stream, _) = listener.accept().unwrap();
640            let mut reader = BufReader::new(stream.try_clone().unwrap());
641            let mut writer = stream;
642            while let Ok((payload, framing)) = read_message(&mut reader) {
643                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
644                let result = match req.method.as_str() {
645                    "initialize" => json!({
646                        "protocolVersion":1,"capabilities":[],"root":root,
647                        "prefix":"Software","generation":2,"revision":10,"identity":"tui"
648                    }),
649                    "issue/ready" | "issue/list" => {
650                        let since = req
651                            .params
652                            .as_ref()
653                            .and_then(|p| p.get("since_revision"))
654                            .cloned();
655                        seen_cb.lock().unwrap().push(since);
656                        json!({
657                            "issues":[{
658                                "id":"atlas-2c3d","state":"TODO","priority":"B",
659                                "title":"Emit a summary table","project":"atlas",
660                                "blocked_by":[],"claimed_by":null,"claimed_at":null
661                            }],
662                            "total":1,"matched":1,"revision":10,
663                            "generation":2,"unchanged":false
664                        })
665                    }
666                    "issue/claim" => json!({
667                        "ok":true,"report":"claimed","issue":null,
668                        "revision":11,"generation":3
669                    }),
670                    other => panic!("unexpected {other}"),
671                };
672                let body = json!({"jsonrpc":"2.0","id":req.id,"result":result});
673                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
674                writer.flush().unwrap();
675            }
676        });
677
678        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
679        let page = backend.ready(None).unwrap();
680        assert_eq!(page.issues[0].id, "atlas-2c3d");
681        assert_eq!(backend.last_since_revision(), Some(None));
682        assert!(backend.claim("atlas-2c3d", false).unwrap().ok);
683        assert_eq!(backend.revision(), 11);
684        backend.ready(None).unwrap();
685        assert_eq!(backend.last_since_revision(), Some(Some(10)));
686        let seen = seen.lock().unwrap();
687        assert_eq!(seen[0], None);
688        assert_eq!(seen[1], Some(json!(10)));
689    }
690
691    #[test]
692    fn root_mismatch_refuses_the_socket() {
693        let dir = tempfile::tempdir().unwrap();
694        let sock = dir.path().join("control.sock");
695        let layout = Layout::new(dir.path().join("vault"), "Software");
696        let listener = UnixListener::bind(&sock).unwrap();
697        thread::spawn(move || {
698            let (stream, _) = listener.accept().unwrap();
699            let mut reader = BufReader::new(stream.try_clone().unwrap());
700            let mut writer = stream;
701            let (payload, framing) = read_message(&mut reader).unwrap();
702            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
703            let body = json!({
704                "jsonrpc": "2.0",
705                "id": req.id,
706                "result": {
707                    "protocolVersion": 1,
708                    "capabilities": [],
709                    "root": "/other/vault",
710                    "prefix": "Software",
711                    "generation": 1,
712                    "revision": 1,
713                    "identity": "tui"
714                }
715            });
716            write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
717            writer.flush().unwrap();
718        });
719        let err = match ControlBackend::connect(&sock, &layout, "tui") {
720            Ok(_) => panic!("expected root mismatch"),
721            Err(err) => err,
722        };
723        match err {
724            ControlAttachError::Mismatch { got_root, .. } => {
725                assert_eq!(got_root, "/other/vault");
726            }
727            other => panic!("{other:?}"),
728        }
729    }
730}
vissue_tui/control.rs

vissue_tui/
control.rs
