Split comenq into put, list, bump, bust, and del subcommands

Cargo.toml

@@ -20,7 +20,6 @@ comenqd = { path = "crates/comenqd", features = ["test-support"] }
 ortho_config = { git = "https://github.com/leynos/ortho-config.git", tag = "v0.4.0" }
 serial_test = "2"
 tempfile = { workspace = true } # latest 3.x at time of writing; update as new patch versions release
-yaque = { workspace = true }
 wiremock = { workspace = true }
 octocrab = { workspace = true }
 test-support = { workspace = true }
@@ -46,7 +45,6 @@ clap = { version = "4.4", features = ["derive", "env"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
   octocrab = "0.38"
-  yaque = "0.6"
   tracing = "0.1"
   tracing-subscriber = { version = "0.3", features = ["env-filter"] }
   anyhow = "1.0"

README.md

@@ -28,10 +28,22 @@ After building, launch the daemon and queue comments with the client:
 ```bash
 make build
 ./target/debug/comenqd &
-./target/debug/comenq owner/repo 123 "Comment body"
+./target/debug/comenq put owner/repo 123 "Comment body"
 ```
 
 Queued requests persist on disk and are posted sequentially by the daemon.
+`put` prints the comment's deterministic eight-character identifier and an
+approximate ETA. By default a fresh comment waits one full cooldown (plus
+its flutter) from enqueue even when the queue is idle; pass `--now` to
+post as soon as the queue allows. The queue can be inspected and
+reordered by identifier:
+
+```bash
+comenq list          # schedule of pending comments with IDs and ETAs
+comenq bump 1a2b3c4d # move to the head of the queue
+comenq bust 1a2b3c4d # move to the tail of the queue
+comenq del 1a2b3c4d  # remove from the queue
+```
 
 ## Running as a user service
 

crates/comenq/src/client.rs

@@ -1,33 +1,46 @@
 //! Client-side communication with the `comenqd` daemon.
 //!
-//! This module contains the logic to serialize a comment request and send it to
-//! the daemon over its Unix Domain Socket. It is separated from `lib.rs` so
-//! that argument parsing remains focused and the network logic is easily
-//! testable.
+//! This module serializes a protocol request, sends it to the daemon over
+//! its Unix Domain Socket, and renders the reply. It is separated from
+//! `lib.rs` so that argument parsing remains focused and the network logic
+//! is easily testable.
 
-use comenq_lib::CommentRequest;
+use comenq_lib::protocol::{Request, Response};
 use std::path::PathBuf;
 use thiserror::Error;
-use tokio::{io::AsyncWriteExt, net::UnixStream};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::UnixStream,
+};
 use tracing::warn;
 
-use crate::Args;
+use crate::output::{render_entry, render_put};
+use crate::{Args, Command};
 
 /// Errors that can occur when interacting with the daemon.
 #[derive(Debug, Error)]
 pub enum ClientError {
     /// Connecting to the daemon failed.
     #[error("failed to connect to daemon: {0}")]
     Connect(#[from] std::io::Error),
-    /// Serializing the request failed.
-    #[error("failed to serialize request: {0}")]
+    /// Serializing the request or parsing the reply failed.
+    #[error("failed to encode or decode a daemon message: {0}")]
     Serialize(#[from] serde_json::Error),
     /// Writing the request to the socket failed.
     #[error("failed to write to daemon: {0}")]
     Write(#[source] std::io::Error),
     /// Shutting down the socket failed.
     #[error("failed to close connection: {0}")]
     Shutdown(#[source] std::io::Error),
+    /// Reading the daemon's reply failed.
+    #[error("failed to read daemon reply: {0}")]
+    Read(#[source] std::io::Error),
+    /// The daemon reported a failure.
+    #[error("daemon refused the request: {0}")]
+    Daemon(String),
+    /// The daemon's reply did not match the request.
+    #[error("unexpected reply from daemon")]
+    UnexpectedResponse,
 }
 
 /// Connect to the first candidate socket that accepts a connection.
@@ -53,83 +66,153 @@ async fn connect_first(candidates: &[PathBuf]) -> Result<UnixStream, ClientError
     })))
 }
 
-/// Send a `CommentRequest` to the daemon.
+/// Send `request` to the first reachable candidate and parse the reply.
+async fn transact(candidates: &[PathBuf], request: &Request) -> Result<Response, ClientError> {
+    let payload = serde_json::to_vec(request)?;
+    let mut stream = connect_first(candidates).await?;
+    stream
+        .write_all(&payload)
+        .await
+        .map_err(ClientError::Write)?;
+    stream.shutdown().await.map_err(ClientError::Shutdown)?;
+    let mut reply = Vec::new();
+    stream
+        .read_to_end(&mut reply)
+        .await
+        .map_err(ClientError::Read)?;
+    Ok(serde_json::from_slice(&reply)?)
+}
+
+/// Execute the parsed command against the daemon and print the outcome.
 ///
 /// # Examples
 ///
 /// ```no_run
-/// # use comenq::{Args, run};
+/// # use comenq::{Args, Command, run};
 /// # use std::path::PathBuf;
-/// # use comenq_lib::DEFAULT_SOCKET_PATH;
 /// # async fn try_run() -> Result<(), comenq::ClientError> {
 /// let args = Args {
-///     repo_slug: "owner/repo".parse().expect("slug"),
-///     pr_number: 1,
-///     comment_body: String::from("Hi"),
-///     socket: Some(PathBuf::from(DEFAULT_SOCKET_PATH)),
+///     socket: Some(PathBuf::from("/tmp/comenq.sock")),
+///     command: Command::List,
 /// };
 /// run(args).await?;
 /// # Ok(())
 /// # }
 /// ```
 pub async fn run(args: Args) -> Result<(), ClientError> {
-    let candidates = args.socket_candidates();
-    let request = CommentRequest {
-        owner: args.repo_slug.owner().to_owned(),
-        repo: args.repo_slug.repo().to_owned(),
-        pr_number: args.pr_number,
-        body: args.comment_body,
+    let request = args.command.to_request();
+    let response = transact(&args.socket_candidates(), &request).await?;
+    let (entry, entries) = match response {
+        Response::Error { message } => return Err(ClientError::Daemon(message)),
+        Response::Ok { entry, entries } => (entry, entries),
     };
-
-    let payload = serde_json::to_vec(&request)?;
-
-    let mut stream = connect_first(&candidates).await?;
-    stream
-        .write_all(&payload)
-        .await
-        .map_err(ClientError::Write)?;
-    if let Err(e) = stream.shutdown().await {
-        warn!("failed to close connection: {e}");
-        return Err(ClientError::Shutdown(e));
+    match &args.command {
+        Command::Put { .. } => {
+            let entry = entry.ok_or(ClientError::UnexpectedResponse)?;
+            println!("{}", render_put(&entry));
+        }
+        Command::List => {
+            let entries = entries.ok_or(ClientError::UnexpectedResponse)?;
+            if entries.is_empty() {
+                println!("No comments queued.");
+            } else {
+                for entry in &entries {
+                    println!("{}", render_entry(entry));
+                }
+            }
+        }
+        Command::Bump { id } => println!("Moved {id} to the head of the queue."),
+        Command::Bust { id } => println!("Moved {id} to the tail of the queue."),
+        Command::Del { id } => println!("Removed {id} from the queue."),
     }
     Ok(())
 }
 
 #[cfg(test)]
 mod tests {
+    //! Round-trip tests for the client transport.
     use super::{ClientError, run};
-    use crate::{Args, RepoSlug};
-    use comenq_lib::CommentRequest;
+    use crate::{Args, Command};
+    use comenq_lib::protocol::{PendingEntry, Request, Response};
     use tempfile::tempdir;
-    use tokio::io::AsyncReadExt;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::net::UnixListener;
 
-    #[tokio::test]
-    async fn run_sends_request() {
-        let dir = tempdir().expect("temp dir");
-        let socket = dir.path().join("sock");
-        let listener = UnixListener::bind(&socket).expect("bind socket");
+    fn put_args(socket: std::path::PathBuf) -> Args {
+        Args {
+            socket: Some(socket),
+            command: Command::Put {
+                repo_slug: "octocat/hello-world".parse().expect("slug"),
+                pr_number: 1,
+                comment_body: "Hi".into(),
+                now: false,
+            },
+        }
+    }
 
-        let accept = tokio::spawn(async move {
+    /// Accept one connection, capture the request, and reply.
+    fn spawn_daemon(listener: UnixListener, reply: Response) -> tokio::task::JoinHandle<Request> {
+        tokio::spawn(async move {
             let (mut stream, _) = listener.accept().await.expect("accept");
             let mut buf = Vec::new();
             stream.read_to_end(&mut buf).await.expect("read");
-            serde_json::from_slice::<CommentRequest>(&buf).expect("deserialize")
-        });
+            let request = serde_json::from_slice::<Request>(&buf).expect("deserialize");
+            let bytes = serde_json::to_vec(&reply).expect("serialize reply");
+            stream.write_all(&bytes).await.expect("write reply");
+            request
+        })
+    }
 
-        let args = Args {
-            repo_slug: "octocat/hello-world".parse().expect("slug"),
+    #[tokio::test]
+    async fn run_sends_put_request_and_accepts_reply() {
+        let dir = tempdir().expect("temp dir");
+        let socket = dir.path().join("sock");
+        let listener = UnixListener::bind(&socket).expect("bind socket");
+        let reply = Response::entry(PendingEntry {
+            id: "1a2b3c4d".into(),
+            eta_seconds: 0,
+            owner: "octocat".into(),
+            repo: "hello-world".into(),
             pr_number: 1,
-            comment_body: "Hi".into(),
-            socket: Some(socket.clone()),
+            body: "Hi".into(),
+        });
+        let accept = spawn_daemon(listener, reply);
+
+        run(put_args(socket)).await.expect("run succeeds");
+        let request = accept.await.expect("join");
+        let Request::Put { request, immediate } = request else {
+            panic!("expected put request, got {request:?}");
         };
+        assert_eq!(request.owner, "octocat");
+        assert_eq!(request.repo, "hello-world");
+        assert_eq!(request.pr_number, 1);
+        assert_eq!(request.body, "Hi");
+        assert!(!immediate, "put must default to deferred posting");
+    }
+
+    #[tokio::test]
+    async fn run_surfaces_daemon_errors() {
+        let dir = tempdir().expect("temp dir");
+        let socket = dir.path().join("sock");
+        let listener = UnixListener::bind(&socket).expect("bind socket");
+        let accept = spawn_daemon(listener, Response::error("queue unavailable"));
 
-        run(args).await.expect("run succeeds");
-        let req = accept.await.expect("join");
-        assert_eq!(req.owner, "octocat");
-        assert_eq!(req.repo, "hello-world");
-        assert_eq!(req.pr_number, 1);
-        assert_eq!(req.body, "Hi");
+        let err = run(put_args(socket)).await.expect_err("should error");
+        assert!(matches!(err, ClientError::Daemon(m) if m == "queue unavailable"));
+        accept.await.expect("join");
+    }
+
+    #[tokio::test]
+    async fn run_rejects_mismatched_reply() {
+        let dir = tempdir().expect("temp dir");
+        let socket = dir.path().join("sock");
+        let listener = UnixListener::bind(&socket).expect("bind socket");
+        // A bare Ok reply lacks the entry a put expects.
+        let accept = spawn_daemon(listener, Response::ok());
+
+        let err = run(put_args(socket)).await.expect_err("should error");
+        assert!(matches!(err, ClientError::UnexpectedResponse));
+        accept.await.expect("join");
     }
 
     /// A stale socket file (bound once, listener dropped, file left behind)
@@ -170,21 +253,7 @@ mod tests {
         let dir = tempdir().expect("temp dir");
         let socket = dir.path().join("nosock");
 
-        let args = Args {
-            repo_slug: "octocat/hello-world".parse().expect("slug"),
-            pr_number: 1,
-            comment_body: "Hi".into(),
-            socket: Some(socket.clone()),
-        };
-
-        let err = run(args).await.expect_err("should error");
+        let err = run(put_args(socket)).await.expect_err("should error");
         assert!(matches!(err, ClientError::Connect(_)));
     }
-
-    #[test]
-    fn slug_parses() {
-        let slug: RepoSlug = "octocat/hello-world".parse().expect("slug");
-        assert_eq!(slug.owner(), "octocat");
-        assert_eq!(slug.repo(), "hello-world");
-    }
 }