ljos_cli/
note_queue.rs

1//! Memory copies of `ljos note` that the pack did not take.
2//!
3//! `ljos note` writes the logbook line first, then a `note` atom to the
4//! pack. When the pack is off or not answering, the atom waits here, one
5//! JSON line per note, and the next note that reaches the pack, `ljos
6//! finish`, or `ljos doctor` sends it. A refusal is not queued: the pack
7//! would refuse it again. The queue holds one entry per issue and text,
8//! and packsetd answers an identical issue and text with the claim it
9//! already holds, so a resend after a lost answer stores nothing new.
10
11use std::io::Write;
12use std::path::{Path, PathBuf};
13
14use anyhow::{Context, Result};
15use serde::{Deserialize, Serialize};
16
17/// One note waiting for the pack.
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub struct Queued {
20    pub issue: String,
21    pub text: String,
22    /// The pack workspace the note was meant for.
23    pub workspace: String,
24    /// Unix seconds when the pack did not take it.
25    pub at: u64,
26}
27
28impl Queued {
29    fn same(&self, other: &Self) -> bool {
30        self.issue.trim() == other.issue.trim()
31            && self.text.trim() == other.text.trim()
32            && self.workspace == other.workspace
33    }
34}
35
36/// What one send says about a queued note.
37pub enum Sent {
38    /// The pack holds it, under this id.
39    Kept(String),
40    /// The pack refused it; the reason. The note leaves the queue.
41    Refused(String),
42    /// The pack did not answer; the note and every later one wait.
43    Wait(String),
44}
45
46/// What a drain did.
47#[derive(Debug, Default, Clone, PartialEq, Eq)]
48pub struct Drained {
49    /// Issue and pack id of each note the pack now holds.
50    pub kept: Vec<(String, String)>,
51    /// Issue and reason of each note the pack refused.
52    pub refused: Vec<(String, String)>,
53    /// Notes still waiting.
54    pub left: usize,
55    /// Why the first waiting note did not go, when one did not.
56    pub waiting_on: Option<String>,
57}
58
59/// The state dir that holds the queue: `$XDG_STATE_HOME/ljos`, else
60/// `~/.local/state/ljos`.
61#[must_use]
62pub fn state_dir() -> PathBuf {
63    std::env::var_os("XDG_STATE_HOME")
64        .map(PathBuf::from)
65        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
66        .unwrap_or_else(|| PathBuf::from(".local/state"))
67        .join("ljos")
68}
69
70fn queue_file(dir: &Path) -> PathBuf {
71    dir.join("note-queue.jsonl")
72}
73
74fn lock(dir: &Path) -> crate::CommitLock {
75    let _ = std::fs::create_dir_all(dir);
76    crate::CommitLock::acquire(&dir.join("note-queue.lock"))
77}
78
79fn read(dir: &Path) -> Vec<Queued> {
80    std::fs::read_to_string(queue_file(dir))
81        .unwrap_or_default()
82        .lines()
83        .filter_map(|line| serde_json::from_str(line).ok())
84        .collect()
85}
86
87fn write(dir: &Path, notes: &[Queued]) -> Result<()> {
88    let path = queue_file(dir);
89    if notes.is_empty() {
90        if path.exists() {
91            std::fs::write(&path, "").with_context(|| format!("empty {}", path.display()))?;
92        }
93        return Ok(());
94    }
95    let tmp = dir.join("note-queue.jsonl.tmp");
96    let mut body = String::new();
97    for note in notes {
98        body.push_str(&serde_json::to_string(note)?);
99        body.push('\n');
100    }
101    std::fs::write(&tmp, body).with_context(|| format!("write {}", tmp.display()))?;
102    std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))
103}
104
105/// The notes waiting in `dir`, oldest first.
106#[must_use]
107pub fn pending_in(dir: &Path) -> Vec<Queued> {
108    let _held = lock(dir);
109    read(dir)
110}
111
112/// Add `note` to the queue in `dir`. `false` when the same issue and
113/// text already wait there.
114///
115/// # Errors
116///
117/// The queue file cannot be written.
118pub fn push_in(dir: &Path, note: Queued) -> Result<bool> {
119    let _held = lock(dir);
120    if read(dir).iter().any(|seen| seen.same(&note)) {
121        return Ok(false);
122    }
123    let path = queue_file(dir);
124    let mut file = std::fs::OpenOptions::new()
125        .create(true)
126        .append(true)
127        .open(&path)
128        .with_context(|| format!("open {}", path.display()))?;
129    writeln!(file, "{}", serde_json::to_string(&note)?)
130        .with_context(|| format!("append {}", path.display()))?;
131    Ok(true)
132}
133
134/// Send each waiting note in `dir` through `send`, oldest first. A kept
135/// or refused note leaves the queue; the first one that has to wait stops
136/// the drain, and it and every later note stay, in order.
137///
138/// # Errors
139///
140/// The queue file cannot be rewritten.
141pub fn drain_in(dir: &Path, mut send: impl FnMut(&Queued) -> Sent) -> Result<Drained> {
142    let _held = lock(dir);
143    let notes = read(dir);
144    let mut done = Drained::default();
145    if notes.is_empty() {
146        return Ok(done);
147    }
148    let mut rest = Vec::new();
149    for (at, note) in notes.iter().enumerate() {
150        match send(note) {
151            Sent::Kept(id) => done.kept.push((note.issue.clone(), id)),
152            Sent::Refused(why) => done.refused.push((note.issue.clone(), why)),
153            Sent::Wait(why) => {
154                done.waiting_on = Some(why);
155                rest.extend_from_slice(&notes[at..]);
156                break;
157            }
158        }
159    }
160    done.left = rest.len();
161    write(dir, &rest)?;
162    Ok(done)
163}
164
165/// The lines a drain prints: what went, what the pack refused, and what
166/// still waits. Empty when nothing was queued.
167#[must_use]
168pub fn drained_lines(done: &Drained) -> String {
169    let mut out = String::new();
170    if !done.kept.is_empty() {
171        let ids: Vec<String> = done
172            .kept
173            .iter()
174            .map(|(issue, id)| format!("{issue} {}", id.chars().take(8).collect::<String>()))
175            .collect();
176        out.push_str(&format!(
177            "memory: sent {} queued note(s): {}\n",
178            done.kept.len(),
179            ids.join(", ")
180        ));
181    }
182    for (issue, why) in &done.refused {
183        out.push_str(&format!(
184            "memory: the pack refused the queued note on {issue} ({why}); it stays on the logbook\n"
185        ));
186    }
187    if done.left > 0 {
188        out.push_str(&format!(
189            "memory: {} note(s) still queued ({})\n",
190            done.left,
191            done.waiting_on
192                .as_deref()
193                .unwrap_or("the pack did not answer")
194        ));
195    }
196    out
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    fn note(issue: &str, text: &str) -> Queued {
204        Queued {
205            issue: issue.into(),
206            text: text.into(),
207            workspace: "ws".into(),
208            at: 1,
209        }
210    }
211
212    fn dir(name: &str) -> PathBuf {
213        let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
214            .join("../../target/note-queue")
215            .join(name);
216        let _ = std::fs::remove_dir_all(&dir);
217        std::fs::create_dir_all(&dir).unwrap();
218        dir
219    }
220
221    #[test]
222    fn the_same_note_waits_once() {
223        let dir = dir("once");
224        assert!(push_in(&dir, note("p-1", "one line")).unwrap());
225        assert!(!push_in(&dir, note("p-1", " one line ")).unwrap());
226        assert!(push_in(&dir, note("p-2", "one line")).unwrap());
227        assert_eq!(pending_in(&dir).len(), 2);
228    }
229
230    #[test]
231    fn a_drain_keeps_order_and_stops_at_the_first_wait() {
232        let dir = dir("order");
233        for (issue, text) in [("p-1", "a"), ("p-2", "b"), ("p-3", "c"), ("p-4", "d")] {
234            push_in(&dir, note(issue, text)).unwrap();
235        }
236        let mut seen = Vec::new();
237        let done = drain_in(&dir, |n| {
238            seen.push(n.issue.clone());
239            match n.issue.as_str() {
240                "p-1" => Sent::Kept("id-1".into()),
241                "p-2" => Sent::Refused("too long".into()),
242                _ => Sent::Wait("connection refused".into()),
243            }
244        })
245        .unwrap();
246        assert_eq!(seen, ["p-1", "p-2", "p-3"]);
247        assert_eq!(done.kept, [("p-1".to_string(), "id-1".to_string())]);
248        assert_eq!(done.refused.len(), 1);
249        assert_eq!(done.left, 2);
250        let left: Vec<String> = pending_in(&dir).into_iter().map(|n| n.issue).collect();
251        assert_eq!(left, ["p-3", "p-4"]);
252        let lines = drained_lines(&done);
253        assert!(lines.contains("sent 1 queued note(s): p-1 id-1"), "{lines}");
254        assert!(
255            lines.contains("2 note(s) still queued (connection refused)"),
256            "{lines}"
257        );
258        let done = drain_in(&dir, |_| Sent::Kept("id".into())).unwrap();
259        assert_eq!(done.kept.len(), 2);
260        assert!(pending_in(&dir).is_empty());
261        assert_eq!(
262            drained_lines(&drain_in(&dir, |_| unreachable!()).unwrap()),
263            ""
264        );
265    }
266}
ljos_cli/note_queue.rs

ljos_cli/
note_queue.rs
