1use std::path::{Path, PathBuf};
17
18use anyhow::{Context, Result};
19use serde::{Deserialize, Serialize};
20
21fn default_kind() -> String {
22 "note".into()
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct Queued {
28 pub issue: String,
29 pub text: String,
30 pub workspace: String,
32 pub at: u64,
34 #[serde(default = "default_kind")]
36 pub kind: String,
37 #[serde(default, skip_serializing_if = "Vec::is_empty")]
40 pub supersedes: Vec<String>,
41}
42
43impl Queued {
44 pub(crate) fn same(&self, other: &Self) -> bool {
45 self.issue.trim() == other.issue.trim()
46 && self.text.trim() == other.text.trim()
47 && self.workspace == other.workspace
48 && self.kind == other.kind
49 && self.supersedes == other.supersedes
50 }
51}
52
53pub enum Sent {
55 Kept(String),
57 Refused(String),
59 Wait(String),
61}
62
63#[derive(Debug, Default, Clone, PartialEq, Eq)]
65pub struct Drained {
66 pub kept: Vec<(String, String)>,
68 pub refused: Vec<(String, String)>,
70 pub left: usize,
72 pub waiting_on: Option<String>,
74 pub set_aside: Vec<SetAside>,
76}
77
78#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct SetAside {
83 pub path: PathBuf,
84 pub bytes: usize,
85}
86
87#[must_use]
90pub fn state_dir() -> PathBuf {
91 std::env::var_os("XDG_STATE_HOME")
92 .map(PathBuf::from)
93 .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
94 .unwrap_or_else(|| PathBuf::from(".local/state"))
95 .join("ljos")
96}
97
98fn queue_file(dir: &Path) -> PathBuf {
99 dir.join("note-queue.jsonl")
100}
101
102fn lock_file(dir: &Path) -> PathBuf {
103 dir.join("note-queue.lock")
104}
105
106fn lock(dir: &Path) -> crate::CommitLock {
107 let _ = std::fs::create_dir_all(dir);
108 crate::CommitLock::acquire_private(&lock_file(dir))
109}
110
111#[must_use]
114pub fn busy_in(dir: &Path) -> Option<PathBuf> {
115 let lock = lock_file(dir);
116 crate::CommitLock::is_held_private(&lock).then_some(lock)
117}
118
119fn read(dir: &Path) -> Vec<Queued> {
120 read_split(dir).0
121}
122
123fn read_split(dir: &Path) -> (Vec<Queued>, Vec<u8>) {
128 let bytes = crate::read_state_file(&queue_file(dir)).unwrap_or_default();
129 let mut notes = Vec::new();
130 let mut unread = Vec::new();
131 for line in bytes.split_inclusive(|&b| b == b'\n') {
132 match line.strip_suffix(b"\n").map(serde_json::from_slice) {
133 Some(Ok(note)) => notes.push(note),
134 _ if line == b"\n" => {}
135 _ => unread.extend_from_slice(line),
136 }
137 }
138 (notes, unread)
139}
140
141fn damaged_file(dir: &Path) -> PathBuf {
143 dir.join(format!(
144 "note-queue.jsonl.{}.damaged",
145 chrono::Utc::now().format("%Y-%m-%d")
146 ))
147}
148
149#[must_use]
152pub fn set_aside_in(dir: &Path) -> Vec<(PathBuf, u64)> {
153 match crate::open_state_dir(&queue_file(dir)) {
154 Ok((fd, _)) => set_aside_at(&fd, dir),
155 Err(_) => Vec::new(),
156 }
157}
158
159#[must_use]
166pub fn set_aside_at(fd: &std::fs::File, dir: &Path) -> Vec<(PathBuf, u64)> {
167 let Ok(names) = crate::deedbag::list_names(fd) else {
168 return Vec::new();
169 };
170 let mut found: Vec<(PathBuf, u64)> = names
171 .iter()
172 .filter_map(|name| {
173 let text = name.to_str().ok()?;
174 text.strip_prefix("note-queue.jsonl.")?
175 .strip_suffix(".damaged")?;
176 let st = crate::deedbag::stat_at(fd, name).ok()??;
177 let size = u64::try_from(st.st_size).ok()?;
178 ((st.st_mode & libc::S_IFMT) == libc::S_IFREG).then(|| (dir.join(text), size))
179 })
180 .collect();
181 found.sort();
182 found
183}
184
185fn set_aside(dir: &Path, unread: &[u8]) -> Result<Option<SetAside>> {
188 if unread.is_empty() {
189 return Ok(None);
190 }
191 let path = damaged_file(dir);
192 crate::append_locked_row(&path, unread, crate::Durability::Durable)
193 .and_then(|()| crate::sync_state_file(&path))
194 .with_context(|| format!("set aside unreadable queue bytes in {}", path.display()))?;
195 Ok(Some(SetAside {
196 path,
197 bytes: unread.len(),
198 }))
199}
200
201fn write(dir: &Path, notes: &[Queued], unread: &[u8]) -> Result<Option<SetAside>> {
205 let aside = set_aside(dir, unread)?;
206 let path = queue_file(dir);
207 if notes.is_empty() {
208 let present = crate::open_state_dir(&path)
211 .and_then(|(fd, name)| crate::open_regular_at(&fd, &name, &path, libc::O_RDONLY, 0));
212 match present {
213 Ok(_) => crate::rewrite_state_file(&path, b"")
214 .with_context(|| format!("empty {}", path.display()))?,
215 Err(err) if err.raw_os_error() == Some(libc::ENOENT) => {}
216 Err(err) => return Err(err).with_context(|| format!("empty {}", path.display())),
217 }
218 return Ok(aside);
219 }
220 let mut body = String::new();
221 for note in notes {
222 body.push_str(&serde_json::to_string(note)?);
223 body.push('\n');
224 }
225 crate::rewrite_state_file(&path, body.as_bytes())
226 .with_context(|| format!("replace {}", path.display()))?;
227 Ok(aside)
228}
229
230#[must_use]
232pub fn pending_in(dir: &Path) -> Vec<Queued> {
233 let _held = lock(dir);
234 read(dir)
235}
236
237pub fn replace_in(dir: &Path, notes: Vec<Queued>) -> Result<()> {
243 let _held = lock(dir);
244 let (_, unread) = read_split(dir);
245 write(dir, ¬es, &unread).map(drop)
246}
247
248pub fn tidy_in(
257 dir: &Path,
258 fit: impl FnOnce(Vec<Queued>) -> Vec<Queued>,
259) -> Result<Option<SetAside>> {
260 let _held = lock(dir);
261 let (notes, unread) = read_split(dir);
262 let fitted = fit(notes.clone());
263 if fitted == notes && unread.is_empty() {
264 return Ok(None);
265 }
266 write(dir, &fitted, &unread)
267}
268
269pub fn push_in(dir: &Path, note: Queued) -> Result<bool> {
276 let _held = lock(dir);
277 if read(dir).iter().any(|seen| seen.same(¬e)) {
278 return Ok(false);
279 }
280 let path = queue_file(dir);
281 let mut line = serde_json::to_string(¬e)?;
282 line.push('\n');
283 crate::append_locked_row(&path, line.as_bytes(), crate::Durability::Durable)
284 .with_context(|| format!("append {}", path.display()))?;
285 Ok(true)
286}
287
288pub fn drain_in(dir: &Path, mut send: impl FnMut(&Queued) -> Sent) -> Result<Drained> {
296 let _held = lock(dir);
297 let (notes, unread) = read_split(dir);
298 let mut done = Drained::default();
299 if notes.is_empty() && unread.is_empty() {
300 return Ok(done);
301 }
302 let mut rest = Vec::new();
303 for (at, note) in notes.iter().enumerate() {
304 match send(note) {
305 Sent::Kept(id) => done.kept.push((note.issue.clone(), id)),
306 Sent::Refused(why) => done.refused.push((note.issue.clone(), why)),
307 Sent::Wait(why) => {
308 done.waiting_on = Some(why);
309 rest.extend_from_slice(¬es[at..]);
310 break;
311 }
312 }
313 }
314 done.left = rest.len();
315 done.set_aside.extend(write(dir, &rest, &unread)?);
316 Ok(done)
317}
318
319#[must_use]
322pub fn drained_lines(done: &Drained) -> String {
323 let mut out = String::new();
324 if !done.kept.is_empty() {
325 let ids: Vec<String> = done
326 .kept
327 .iter()
328 .map(|(issue, id)| format!("{issue} {}", id.chars().take(8).collect::<String>()))
329 .collect();
330 out.push_str(&format!(
331 "memory: sent {} queued note(s): {}\n",
332 done.kept.len(),
333 ids.join(", ")
334 ));
335 }
336 for (issue, why) in &done.refused {
337 out.push_str(&format!(
338 "memory: the pack refused the queued note on {issue} ({why}); it stays on the logbook\n"
339 ));
340 }
341 for aside in &done.set_aside {
342 out.push_str(&format!(
343 "memory: moved {} byte(s) of the note queue that are not a note to {}; they were not sent\n",
344 aside.bytes,
345 aside.path.display()
346 ));
347 }
348 if done.left > 0 {
349 out.push_str(&format!(
350 "memory: {} note(s) still queued ({})\n",
351 done.left,
352 done.waiting_on
353 .as_deref()
354 .unwrap_or("the pack did not answer")
355 ));
356 }
357 out
358}
359
360#[cfg(test)]
361mod tests {
362 use super::*;
363
364 fn note(issue: &str, text: &str) -> Queued {
365 Queued {
366 issue: issue.into(),
367 text: text.into(),
368 workspace: "ws".into(),
369 at: 1,
370 kind: "note".into(),
371 supersedes: Vec::new(),
372 }
373 }
374
375 fn dir(name: &str) -> PathBuf {
376 let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
377 .join("../../target/note-queue")
378 .join(name);
379 let _ = std::fs::remove_dir_all(&dir);
380 std::fs::create_dir_all(&dir).unwrap();
381 dir
382 }
383
384 #[test]
385 fn an_old_line_without_kind_is_a_note() {
386 let dir = dir("old");
387 std::fs::write(
388 dir.join("note-queue.jsonl"),
389 "{\"issue\":\"p-1\",\"text\":\"one line\",\"workspace\":\"ws\",\"at\":1}\n",
390 )
391 .unwrap();
392 let pending = pending_in(&dir);
393 assert_eq!(pending.len(), 1);
394 assert_eq!(pending[0].kind, "note");
395 assert_eq!(pending[0].text, "one line");
396 assert!(pending[0].supersedes.is_empty());
397 }
398
399 #[test]
400 fn a_lesson_that_closes_another_atom_is_a_different_copy() {
401 let dir = dir("closes");
402 let mut lesson = note("p-1", "one line");
403 lesson.kind = "lesson".into();
404 lesson.supersedes = vec!["atom-old".into()];
405 assert!(push_in(&dir, lesson.clone()).unwrap());
406 assert!(!push_in(&dir, lesson.clone()).unwrap());
407 let mut other = lesson;
408 other.supersedes = vec!["atom-new".into()];
409 assert!(push_in(&dir, other).unwrap());
410 assert_eq!(pending_in(&dir).len(), 2);
411 }
412
413 #[test]
416 fn a_torn_last_copy_leaves_the_queue() {
417 let dir = dir("torn");
418 let mut bytes =
419 format!("{}\n", serde_json::to_string(¬e("p-1", "kept")).unwrap()).into_bytes();
420 bytes.extend_from_slice(b"{\"issue\":\"p-2\",\"text\":\"caf\xc3");
421 std::fs::write(dir.join("note-queue.jsonl"), bytes).unwrap();
422 let pending = pending_in(&dir);
423 assert_eq!(pending.len(), 1, "{pending:?}");
424 assert_eq!(pending[0].text, "kept");
425 }
426
427 #[test]
431 fn concurrent_queue_rows_each_parse() {
432 use std::os::unix::fs::PermissionsExt;
433 let dir = dir("concurrent");
434 const THREADS: usize = 32;
435 const EACH: usize = 4;
436 let mut handles = Vec::with_capacity(THREADS);
437 for thread in 0..THREADS {
438 let dir = dir.clone();
439 handles.push(std::thread::spawn(move || {
440 for n in 0..EACH {
441 let mut queued = note(&format!("p-{thread}"), &format!("text {thread}-{n}"));
442 queued.at = (thread * EACH + n) as u64;
443 push_in(&dir, queued).unwrap();
444 }
445 }));
446 }
447 for handle in handles {
448 handle.join().expect("queue writer");
449 }
450 let path = dir.join("note-queue.jsonl");
451 let text = std::fs::read_to_string(&path).unwrap();
452 let rows: Vec<Queued> = text
453 .lines()
454 .map(|line| {
455 serde_json::from_str(line).unwrap_or_else(|err| {
456 panic!("queue row does not parse ({err}): {line}");
457 })
458 })
459 .collect();
460 assert_eq!(rows.len(), THREADS * EACH, "{text}");
461 let mode = std::fs::symlink_metadata(&path)
462 .unwrap()
463 .permissions()
464 .mode()
465 & 0o777;
466 assert_eq!(mode, 0o600, "queue mode {mode:o}");
467 }
468
469 #[test]
474 fn a_rewrite_stays_private_and_refuses_a_symlink() {
475 use std::os::unix::fs::PermissionsExt;
476 let dir = dir("rewrite");
477 push_in(&dir, note("p-1", "a")).unwrap();
478 push_in(&dir, note("p-2", "b")).unwrap();
479 let done = drain_in(&dir, |n| match n.issue.as_str() {
480 "p-1" => Sent::Kept("id".into()),
481 _ => Sent::Wait("down".into()),
482 })
483 .unwrap();
484 assert_eq!(done.left, 1);
485 let path = queue_file(&dir);
486 let mode = std::fs::symlink_metadata(&path)
487 .unwrap()
488 .permissions()
489 .mode()
490 & 0o777;
491 assert_eq!(mode, 0o600, "queue mode after rewrite {mode:o}");
492 let lock_mode = std::fs::symlink_metadata(dir.join("note-queue.lock"))
493 .unwrap()
494 .permissions()
495 .mode()
496 & 0o777;
497 assert_eq!(lock_mode, 0o600, "lock mode {lock_mode:o}");
498
499 let outside = dir.join("outside");
500 std::fs::write(&outside, "kept\n").unwrap();
501 std::fs::remove_file(&path).unwrap();
502 std::os::unix::fs::symlink(&outside, &path).unwrap();
503 assert!(replace_in(&dir, vec![note("p-9", "forged")]).is_err());
504 assert_eq!(std::fs::read_to_string(&outside).unwrap(), "kept\n");
505 }
506
507 #[test]
510 fn emptying_a_dangling_linked_queue_is_refused() {
511 let dir = dir("empty-dangling");
512 let path = queue_file(&dir);
513 std::os::unix::fs::symlink(dir.join("gone"), &path).unwrap();
514 assert!(replace_in(&dir, Vec::new()).is_err());
515 assert!(std::fs::symlink_metadata(&path).unwrap().is_symlink());
516 }
517
518 #[test]
523 fn a_drain_sets_aside_what_it_cannot_read() {
524 use std::os::unix::fs::PermissionsExt;
525 let dir = dir("aside");
526 let sent = format!("{}\n", serde_json::to_string(¬e("p-1", "sent")).unwrap());
527 let waits = format!(
528 "{}\n",
529 serde_json::to_string(¬e("p-2", "waits")).unwrap()
530 );
531 let cut: &[u8] = b"{\"issue\":\"p-3\",\"te\n";
532 let tail: &[u8] =
533 br#"{"issue":"p-4","text":"no newline","workspace":"ws","at":1,"kind":"note"}"#;
534 let mut bytes = sent.into_bytes();
535 bytes.extend_from_slice(cut);
536 bytes.extend_from_slice(waits.as_bytes());
537 bytes.extend_from_slice(tail);
538 std::fs::write(queue_file(&dir), &bytes).unwrap();
539 let done = drain_in(&dir, |n| match n.issue.as_str() {
540 "p-1" => Sent::Kept("id".into()),
541 _ => Sent::Wait("down".into()),
542 })
543 .unwrap();
544 assert_eq!(done.left, 1, "{done:?}");
545 assert_eq!(std::fs::read_to_string(queue_file(&dir)).unwrap(), waits);
546 let unread = [cut, tail].concat();
547 let [aside] = done.set_aside.as_slice() else {
548 panic!("{done:?}");
549 };
550 assert_eq!(aside.bytes, unread.len());
551 assert_eq!(std::fs::read(&aside.path).unwrap(), unread);
552 let name = aside.path.file_name().unwrap().to_str().unwrap();
553 assert!(
554 name.starts_with("note-queue.jsonl.") && name.ends_with(".damaged"),
555 "{name}"
556 );
557 let mode = std::fs::metadata(&aside.path).unwrap().permissions().mode() & 0o777;
558 assert_eq!(mode, 0o600, "damaged file mode {mode:o}");
559 let lines = drained_lines(&done);
560 assert!(lines.contains(&aside.path.display().to_string()), "{lines}");
561 }
562
563 #[test]
566 fn a_tidy_sets_aside_a_queue_with_no_note_in_it() {
567 let dir = dir("tidy");
568 std::fs::write(queue_file(&dir), b"{\"issue\":\"p-1\",\"te").unwrap();
569 let aside = tidy_in(&dir, |notes| notes).unwrap().expect("set aside");
570 assert_eq!(
571 std::fs::read(&aside.path).unwrap(),
572 b"{\"issue\":\"p-1\",\"te"
573 );
574 assert_eq!(std::fs::read(queue_file(&dir)).unwrap(), b"");
575 assert_eq!(tidy_in(&dir, |notes| notes).unwrap(), None);
576 }
577
578 #[test]
582 fn a_linked_damaged_name_is_not_followed() {
583 let dir = dir("linked");
584 let real = dir.join("note-queue.jsonl.2026-01-01.damaged");
585 std::fs::write(&real, b"cut").unwrap();
586 let outside = dir.parent().unwrap().join("linked-outside");
587 std::fs::write(&outside, b"not ljos's own").unwrap();
588 std::os::unix::fs::symlink(&outside, dir.join("note-queue.jsonl.2026-01-02.damaged"))
589 .unwrap();
590 std::os::unix::fs::symlink(
591 dir.join("gone"),
592 dir.join("note-queue.jsonl.2026-01-03.damaged"),
593 )
594 .unwrap();
595 assert_eq!(set_aside_in(&dir), [(real, 3)]);
596 assert_eq!(std::fs::read(&outside).unwrap(), b"not ljos's own");
597 assert!(!dir.join("gone").exists());
598 }
599
600 #[test]
603 fn a_linked_state_dir_lists_nothing_from_outside() {
604 let root = dir("linked-dir");
605 let outside = root.join("outside");
606 std::fs::create_dir(&outside).unwrap();
607 std::fs::write(outside.join("note-queue.jsonl.2026-01-01.damaged"), b"x").unwrap();
608 let state = root.join("ljos");
609 std::os::unix::fs::symlink(&outside, &state).unwrap();
610 assert_eq!(set_aside_in(&state), []);
611 }
612
613 #[test]
614 fn the_same_note_waits_once() {
615 let dir = dir("once");
616 assert!(push_in(&dir, note("p-1", "one line")).unwrap());
617 assert!(!push_in(&dir, note("p-1", " one line ")).unwrap());
618 assert!(push_in(&dir, note("p-2", "one line")).unwrap());
619 assert_eq!(pending_in(&dir).len(), 2);
620 }
621
622 #[test]
623 fn a_drain_keeps_order_and_stops_at_the_first_wait() {
624 let dir = dir("order");
625 for (issue, text) in [("p-1", "a"), ("p-2", "b"), ("p-3", "c"), ("p-4", "d")] {
626 push_in(&dir, note(issue, text)).unwrap();
627 }
628 let mut seen = Vec::new();
629 let done = drain_in(&dir, |n| {
630 seen.push(n.issue.clone());
631 match n.issue.as_str() {
632 "p-1" => Sent::Kept("id-1".into()),
633 "p-2" => Sent::Refused("too long".into()),
634 _ => Sent::Wait("connection refused".into()),
635 }
636 })
637 .unwrap();
638 assert_eq!(seen, ["p-1", "p-2", "p-3"]);
639 assert_eq!(done.kept, [("p-1".to_string(), "id-1".to_string())]);
640 assert_eq!(done.refused.len(), 1);
641 assert_eq!(done.left, 2);
642 let left: Vec<String> = pending_in(&dir).into_iter().map(|n| n.issue).collect();
643 assert_eq!(left, ["p-3", "p-4"]);
644 let lines = drained_lines(&done);
645 assert!(lines.contains("sent 1 queued note(s): p-1 id-1"), "{lines}");
646 assert!(
647 lines.contains("2 note(s) still queued (connection refused)"),
648 "{lines}"
649 );
650 let done = drain_in(&dir, |_| Sent::Kept("id".into())).unwrap();
651 assert_eq!(done.kept.len(), 2);
652 assert!(pending_in(&dir).is_empty());
653 assert_eq!(
654 drained_lines(&drain_in(&dir, |_| unreachable!()).unwrap()),
655 ""
656 );
657 }
658}