feat(clipcatctl): add tail subcommand to follow clipboard history

Cargo.toml

@@ -56,7 +56,7 @@ tokio = { version = "1", features = [
   "sync",
   "process",
 ] }
-tokio-stream = { version = "0.1", features = ["net"] }
+tokio-stream = { version = "0.1", features = ["net", "sync"] }
 
 arboard = { version = "3", default-features = false, features = [
   "image-data",
@@ -90,6 +90,7 @@ humansize = "2"
 image = "0.25"
 libc = "0.2"
 linicon = "2"
+lru = "0.16"
 mime = "0.3"
 mio = { version = "1.2", features = ["os-ext"] }
 notify = "8"

README.md

@@ -165,12 +165,13 @@ clipcatd --no-daemon
 
 3. You can run the following commands with `clipcatctl` or `clipcat-menu`:
 
-| Command                   | Comment                                               |
-| ------------------------- | ----------------------------------------------------- |
-| `clipcatctl list`         | List cached clipboard history                         |
-| `clipcatctl promote <id>` | Insert cached clip with `<id>` into the X11 clipboard |
-| `clipcatctl remove [ids]` | Remove cached clips with `[ids]` from the server      |
-| `clipcatctl clear`        | Clear cached clipboard history                        |
+| Command                   | Comment                                                            |
+| ------------------------- | ------------------------------------------------------------------ |
+| `clipcatctl list`         | List cached clipboard history                                      |
+| `clipcatctl tail [-f]`    | Print the most recent entries (default 10); `-f` streams new clips |
+| `clipcatctl promote <id>` | Insert cached clip with `<id>` into the X11 clipboard              |
+| `clipcatctl remove [ids]` | Remove cached clips with `[ids]` from the server                   |
+| `clipcatctl clear`        | Clear cached clipboard history                                     |
 
 | Command               | Comment                                     |
 | --------------------- | ------------------------------------------- |
@@ -584,6 +585,53 @@ pkill clipcatd
 
 </details>
 
+<details>
+    <summary>Reacting to new clips with <code>clipcatctl tail</code></summary>
+
+`clipcatctl tail -f` streams each new clip's id and preview as it is
+added, in the same format as `clipcatctl list`. This lets you write
+small shell hooks that mutate clips out-of-band.
+
+Run **one** hook script for every transform you want, even unrelated
+ones. Independent `tail -f` listeners each see the other listeners'
+outputs as new events, so two separate hooks can cascade-trigger each
+other indefinitely. A single script applies all transforms in one pass
+and records the result in a shared state file:
+
+```bash
+#!/usr/bin/env bash
+# `-n 0` → react only to clips added after the hook starts.
+SEEN=~/.cache/clipcat-hooks/produced-ids
+mkdir -p "$(dirname "$SEEN")" && touch "$SEEN"
+
+clipcatctl tail -f -n 0 | while IFS= read -r line; do
+    id=${line%%:*}
+    grep -qxF "$id" "$SEEN" && continue
+    raw=$(clipcatctl get "$id") || continue
+
+    # Apply your transforms here (compose as many as you need).
+    out=${raw#"${raw%%[![:space:]]*}"} # trim leading whitespace
+    out=${out%"${out##*[![:space:]]}"} # trim trailing whitespace
+    # out="[$(date -I)] $out"          # uncomment to date-stamp
+
+    [ "$raw" = "$out" ] && continue
+    new_id=$(clipcatctl update "$id" "$out") || continue
+    printf '%s\n' "$new_id" >>"$SEEN"
+
+    # Cap the state file so it does not grow without bound.
+    if [ "$(wc -l <"$SEEN")" -gt 1024 ]; then
+        tail -n 512 "$SEEN" >"$SEEN.tmp" && mv "$SEEN.tmp" "$SEEN"
+    fi
+done
+```
+
+`clipcatctl update` prints the new clip's id, which the hook records in
+`$SEEN` so it's skipped when `tail -f` re-emits it. `clipcatctl tail -f`
+also reconnects automatically on `clipcatd` restarts and resumes
+streaming new clips, so the script can run as a long-lived service.
+
+</details>
+
 <details>
     <summary>Starting <b>clipcatd</b> with <a href="https://systemd.io/" target="_blank">systemd</a></summary>
 

clipcatctl/Cargo.toml

@@ -27,7 +27,9 @@ bytes         = { workspace = true }
 clap          = { workspace = true }
 clap_complete = { workspace = true }
 directories   = { workspace = true }
+futures       = { workspace = true }
 http          = { workspace = true }
+lru           = { workspace = true }
 mime          = { workspace = true }
 shadow-rs     = { workspace = true }
 shellexpand   = { workspace = true }

clipcatctl/src/cli.rs

@@ -1,9 +1,16 @@
-use std::{io::Write, num::ParseIntError, path::PathBuf};
+use std::{
+    io::Write,
+    num::{NonZeroUsize, ParseIntError},
+    path::PathBuf,
+    time::Duration,
+};
 
 use clap::{CommandFactory, Parser, Subcommand};
 use clipcat_base::{ClipEntryMetadata, ClipboardKind, ClipboardWatcherState};
 use clipcat_client::{Client, History, Manager as _, System, Watcher as _};
 use clipcat_external_editor::ExternalEditor;
+use futures::StreamExt;
+use lru::LruCache;
 use snafu::ResultExt;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
@@ -126,6 +133,27 @@ pub enum Commands {
         no_id: bool,
     },
 
+    #[clap(about = "Print most recent clipboard entries, oldest first; `-f` to follow")]
+    Tail {
+        #[clap(long)]
+        no_id: bool,
+
+        #[clap(
+            long = "lines",
+            short = 'n',
+            default_value = "10",
+            help = "Print the most recent N entries (oldest of those first)"
+        )]
+        lines: u64,
+
+        #[clap(
+            long = "follow",
+            short = 'f',
+            help = "Follow new clipboard entries; reconnects on clipcatd restart"
+        )]
+        follow: bool,
+    },
+
     #[clap(about = "Update clip with <id>")]
     Update {
         #[clap(value_parser = parse_hex)]
@@ -214,6 +242,13 @@ impl Cli {
     pub fn run(self) -> Result<i32, Error> {
         let client_version = Self::command().get_version().unwrap_or_default().to_string();
         match self.commands {
+            Some(Commands::Tail { no_id, lines, follow }) => {
+                let config = self.load_config();
+                config.log.registry();
+                return Runtime::new()
+                    .context(error::InitializeTokioRuntimeSnafu)?
+                    .block_on(run_tail(config, no_id, lines, follow));
+            }
             Some(Commands::Version { client }) if client => {
                 std::io::stdout()
                     .write_all(Self::command().render_long_version().as_bytes())
@@ -466,6 +501,16 @@ fn print_watcher_state(state: ClipboardWatcherState) {
     println!("{msg}");
 }
 
+fn format_metadata_line(
+    metadata: &ClipEntryMetadata,
+    no_id: bool,
+    show_source_prefix: bool,
+) -> String {
+    let ClipEntryMetadata { id, preview, kind, .. } = metadata;
+    let prefix = if show_source_prefix { format!("{} ", kind.prefix()) } else { String::new() };
+    if no_id { format!("{prefix}{preview}\n") } else { format!("{id:016x}: {prefix}{preview}\n") }
+}
+
 async fn print_list(
     client: &Client,
     preview_length: usize,
@@ -474,17 +519,149 @@ async fn print_list(
 ) -> Result<(), Error> {
     let metadata_list = client.list(preview_length).await?;
     for metadata in metadata_list {
-        let ClipEntryMetadata { id, preview, kind, .. } = metadata;
-        let prefix = if show_source_prefix { format!("{} ", kind.prefix()) } else { String::new() };
-        let output = if no_id {
-            format!("{prefix}{preview}\n")
-        } else {
-            format!("{id:016x}: {prefix}{preview}\n")
-        };
-        tokio::io::stdout().write_all(output.as_bytes()).await.context(error::WriteStdoutSnafu)?;
+        let line = format_metadata_line(&metadata, no_id, show_source_prefix);
+        tokio::io::stdout().write_all(line.as_bytes()).await.context(error::WriteStdoutSnafu)?;
     }
     Ok(())
 }
 
 #[inline]
 const fn parse_hex(src: &str) -> Result<u64, ParseIntError> { u64::from_str_radix(src, 16) }
+
+const TAIL_BACKOFF_INITIAL: Duration = Duration::from_millis(100);
+const TAIL_BACKOFF_MAX: Duration = Duration::from_secs(5);
+const TAIL_PRINTED_IDS_CAP: NonZeroUsize = NonZeroUsize::new(1024).unwrap();
+
+struct TailState {
+    printed_ids: LruCache<u64, ()>,
+    is_first_session: bool,
+    backoff: Duration,
+}
+
+impl TailState {
+    fn new() -> Self {
+        Self {
+            printed_ids: LruCache::new(TAIL_PRINTED_IDS_CAP),
+            is_first_session: true,
+            backoff: TAIL_BACKOFF_INITIAL,
+        }
+    }
+
+    fn next_backoff(&mut self) -> Duration {
+        let current = self.backoff;
+        self.backoff = std::cmp::min(self.backoff.saturating_mul(2), TAIL_BACKOFF_MAX);
+        current
+    }
+
+    const fn reset_backoff(&mut self) { self.backoff = TAIL_BACKOFF_INITIAL; }
+}
+
+async fn write_metadata_line(
+    metadata: &ClipEntryMetadata,
+    no_id: bool,
+    show_source_prefix: bool,
+) -> Result<(), Error> {
+    let line = format_metadata_line(metadata, no_id, show_source_prefix);
+    tokio::io::stdout().write_all(line.as_bytes()).await.context(error::WriteStdoutSnafu)?;
+    Ok(())
+}
+
+async fn run_tail(config: Config, no_id: bool, lines: u64, follow: bool) -> Result<i32, Error> {
+    let preview_length = config.preview_length;
+    let show_source_prefix = config.show_source_prefix;
+    let mut state = TailState::new();
+    if !follow {
+        // Single-shot: fail fast if the daemon is unreachable.
+        return run_tail_session(
+            &config,
+            no_id,
+            lines,
+            preview_length,
+            show_source_prefix,
+            &mut state,
+            false,
+        )
+        .await
+        .map(|()| 0);
+    }
+    loop {
+        match run_tail_session(
+            &config,
+            no_id,
+            lines,
+            preview_length,
+            show_source_prefix,
+            &mut state,
+            true,
+        )
+        .await
+        {
+            Ok(()) => return Ok(0),
+            Err(err) => {
+                let delay = state.next_backoff();
+                tracing::debug!(
+                    "tail session ended ({err}); reconnecting in {} ms",
+                    delay.as_millis()
+                );
+                tokio::time::sleep(delay).await;
+            }
+        }
+    }
+}
+
+async fn run_tail_session(
+    config: &Config,
+    no_id: bool,
+    lines: u64,
+    preview_length: usize,
+    show_source_prefix: bool,
+    state: &mut TailState,
+    follow: bool,
+) -> Result<(), Error> {
+    let client = Client::builder()
+        .grpc_endpoint(config.server_endpoint.clone())
+        .access_token(config.access_token())
+        .max_decoding_message_size(config.grpc_max_message_size)
+        .build()
+        .await?;
+
+    // Subscribe before listing so events from the gap between the two are
+    // buffered in the receiver and deduped via `printed_ids`. On reconnect we
+    // skip the list entirely; only entries arriving on the new stream are
+    // emitted.
+    let stream = if follow { Some(client.subscribe(preview_length).await?) } else { None };
+
+    if state.is_first_session {
+        let take = usize::try_from(lines).unwrap_or(0);
+        if take > 0 {
+            let snapshot = client.list(preview_length).await?;
+            // `client.list` returns newest first; take the most recent `take`
+            // entries, then iterate in reverse so output reads oldest -> newest,
+            // matching the chronological order of streamed events under `-f`.
+            let recent: Vec<_> = snapshot.into_iter().take(take).collect();
+            for metadata in recent.iter().rev() {
+                if state.printed_ids.contains(&metadata.id) {
+                    continue;
+                }
+                write_metadata_line(metadata, no_id, show_source_prefix).await?;
+                let _ = state.printed_ids.put(metadata.id, ());
+            }
+        }
+        state.is_first_session = false;
+    }
+
+    let Some(mut stream) = stream else {
+        return Ok(());
+    };
+    while let Some(item) = stream.next().await {
+        let metadata = item?;
+        if state.printed_ids.contains(&metadata.id) {
+            continue;
+        }
+        write_metadata_line(&metadata, no_id, show_source_prefix).await?;
+        let _ = state.printed_ids.put(metadata.id, ());
+        state.reset_backoff();
+    }
+
+    Err(Error::Operation { error: "subscription stream ended".to_owned() })
+}

clipcatctl/src/error.rs

@@ -102,6 +102,12 @@ impl From<clipcat_client::error::ListClipError> for Error {
     }
 }
 
+impl From<clipcat_client::error::SubscribeClipError> for Error {
+    fn from(err: clipcat_client::error::SubscribeClipError) -> Self {
+        Self::Operation { error: err.to_string() }
+    }
+}
+
 impl From<clipcat_client::error::EnableWatcherError> for Error {
     fn from(err: clipcat_client::error::EnableWatcherError) -> Self {
         Self::Operation { error: err.to_string() }

crates/client/Cargo.toml

@@ -15,6 +15,7 @@ keywords.workspace     = true
 [dependencies]
 tracing = { workspace = true }
 
+futures    = { workspace = true }
 hyper-util = { workspace = true }
 tokio      = { workspace = true }