feat: CLI watch mode, batch mode, and proto-ts stub

Makefile

@@ -1,7 +1,7 @@
 .PHONY: build release test fmt fmt-check clippy check clean docker-build docker-run \
        release-patch release-minor release-major github-release \
        publish publish-dry-run bench bench-core bench-protocol bench-compare bench-quick \
-       helm-lint helm-template proto-gen proto-go proto-py \
+       helm-lint helm-template proto-gen proto-go proto-py proto-ts \
        cluster cluster-stop cluster-status cluster-clean
 
 # extract the workspace version from the root Cargo.toml
@@ -154,10 +154,15 @@ proto-go:
 proto-py:
 	cd clients/ember-py && $(MAKE) proto-gen
 
+proto-ts:
+	@echo "TypeScript gRPC client codegen: run the following in clients/ember-ts/"
+	@echo "  npx @grpc/proto-loader-gen-types proto/ember/v1/ember.proto --outDir=src/generated"
+
 # --- helm ---
 
 helm-lint:
 	helm lint helm/ember
 
 helm-template:
 	helm template ember helm/ember
+

crates/ember-cli/src/batch.rs

@@ -0,0 +1,100 @@
+//! Batch mode: reads commands from stdin line by line and executes them.
+//!
+//! Useful for scripting and CI pipelines:
+//!
+//!   echo "SET mykey hello" | ember-cli
+//!   ember-cli < commands.txt
+//!
+//! Blank lines and lines beginning with `#` are treated as comments and
+//! skipped. Each non-empty line is split on whitespace and sent as a RESP3
+//! command. Results are printed to stdout; errors go to stderr.
+
+use std::io::{self, BufRead};
+use std::process::ExitCode;
+
+use colored::Colorize;
+
+use crate::connection::Connection;
+use crate::format::format_response;
+use crate::tls::TlsClientConfig;
+
+/// Reads commands from stdin line by line and executes them against the server.
+///
+/// Exits with failure if any command cannot be sent (i.e. the connection
+/// breaks). Server-level errors (e.g. wrong type, wrong arity) are printed
+/// to stderr but do not stop processing.
+pub fn run_batch(
+    host: &str,
+    port: u16,
+    password: Option<&str>,
+    tls: Option<&TlsClientConfig>,
+) -> ExitCode {
+    let rt = match tokio::runtime::Runtime::new() {
+        Ok(rt) => rt,
+        Err(e) => {
+            eprintln!("{}", format!("failed to create runtime: {e}").red());
+            return ExitCode::FAILURE;
+        }
+    };
+
+    rt.block_on(run_batch_async(host, port, password, tls))
+}
+
+async fn run_batch_async(
+    host: &str,
+    port: u16,
+    password: Option<&str>,
+    tls: Option<&TlsClientConfig>,
+) -> ExitCode {
+    let mut conn = match Connection::connect(host, port, tls).await {
+        Ok(c) => c,
+        Err(e) => {
+            eprintln!("{}", format!("could not connect to {host}:{port}: {e}").red());
+            return ExitCode::FAILURE;
+        }
+    };
+
+    if let Some(pw) = password {
+        if let Err(e) = conn.authenticate(pw).await {
+            eprintln!("{}", format!("authentication failed: {e}").red());
+            conn.shutdown().await;
+            return ExitCode::FAILURE;
+        }
+    }
+
+    let stdin = io::stdin();
+    for line in stdin.lock().lines() {
+        let line = match line {
+            Ok(l) => l,
+            Err(e) => {
+                eprintln!("{}", format!("read error: {e}").red());
+                conn.shutdown().await;
+                return ExitCode::FAILURE;
+            }
+        };
+
+        let trimmed = line.trim();
+
+        // skip blank lines and comments
+        if trimmed.is_empty() || trimmed.starts_with('#') {
+            continue;
+        }
+
+        let tokens: Vec<String> = trimmed
+            .split_whitespace()
+            .map(|s| s.to_string())
+            .collect();
+
+        match conn.send_command(&tokens).await {
+            Ok(frame) => println!("{}", format_response(&frame)),
+            Err(e) => {
+                eprintln!("{}", format!("error: {e}").red());
+                conn.shutdown().await;
+                return ExitCode::FAILURE;
+            }
+        }
+    }
+
+    conn.shutdown().await;
+    ExitCode::SUCCESS
+}

crates/ember-cli/src/main.rs

@@ -2,8 +2,10 @@
 //!
 //! Connects to an ember server over TCP (or TLS), sends commands as RESP3
 //! frames, and pretty-prints responses. Supports one-shot mode, interactive
-//! REPL, and named subcommands for cluster management and benchmarking.
+//! REPL, batch/pipe mode, watch mode, and named subcommands for cluster
+//! management and benchmarking.
 
+mod batch;
 mod bench_conn;
 mod benchmark;
 mod cluster;
@@ -12,6 +14,7 @@ mod connection;
 mod format;
 mod repl;
 mod tls;
+mod watch;
 
 use std::ffi::OsString;
 use std::process::ExitCode;
@@ -66,6 +69,17 @@ enum Mode {
     /// Run a built-in benchmark against the server.
     Benchmark(benchmark::BenchmarkArgs),
 
+    /// Watch a key and print its value whenever it changes.
+    ///
+    /// Polls the server at the given interval. Press ctrl-c to stop.
+    Watch {
+        /// Key to monitor.
+        key: String,
+        /// Poll interval in milliseconds.
+        #[arg(default_value_t = 1000)]
+        interval_ms: u64,
+    },
+
     /// One-shot mode: pass a raw command (e.g. `ember-cli SET key value`).
     #[command(external_subcommand)]
     Raw(Vec<OsString>),
@@ -92,15 +106,33 @@ fn main() -> ExitCode {
 
     match args.mode {
         None => {
-            // interactive REPL mode
-            repl::run_repl(
-                &args.host,
-                args.port,
-                args.password.as_deref(),
-                tls.as_ref(),
-            );
-            ExitCode::SUCCESS
+            if is_stdin_tty() {
+                // interactive REPL mode
+                repl::run_repl(
+                    &args.host,
+                    args.port,
+                    args.password.as_deref(),
+                    tls.as_ref(),
+                );
+                ExitCode::SUCCESS
+            } else {
+                // batch mode: stdin is a pipe or redirected file
+                batch::run_batch(
+                    &args.host,
+                    args.port,
+                    args.password.as_deref(),
+                    tls.as_ref(),
+                )
+            }
         }
+        Some(Mode::Watch { key, interval_ms }) => watch::run_watch(
+            &args.host,
+            args.port,
+            args.password.as_deref(),
+            tls.as_ref(),
+            &key,
+            interval_ms,
+        ),
         Some(Mode::Cluster { cmd }) => cluster::run_cluster(
             &cmd,
             &args.host,
@@ -131,6 +163,15 @@ fn main() -> ExitCode {
     }
 }
 
+/// Returns `true` when stdin is connected to an interactive terminal.
+///
+/// When stdin is a pipe or redirected file this returns `false`, which
+/// triggers batch mode.
+fn is_stdin_tty() -> bool {
+    use std::io::IsTerminal;
+    std::io::stdin().is_terminal()
+}
+
 /// Sends a single command and prints the response.
 fn run_oneshot(
     host: &str,

crates/ember-cli/src/watch.rs

@@ -0,0 +1,113 @@
+//! Watch mode: polls a key at a configurable interval and prints its value
+//! whenever it changes.
+//!
+//! Run until ctrl-c is pressed:
+//!   ember-cli watch mykey
+//!   ember-cli watch mykey 500
+
+use std::process::ExitCode;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use colored::Colorize;
+
+use crate::connection::Connection;
+use crate::format::format_response;
+use crate::tls::TlsClientConfig;
+
+/// Polls `key` every `interval_ms` milliseconds and prints its value when
+/// it changes. Runs until ctrl-c is pressed or the connection fails.
+pub fn run_watch(
+    host: &str,
+    port: u16,
+    password: Option<&str>,
+    tls: Option<&TlsClientConfig>,
+    key: &str,
+    interval_ms: u64,
+) -> ExitCode {
+    let rt = match tokio::runtime::Runtime::new() {
+        Ok(rt) => rt,
+        Err(e) => {
+            eprintln!("{}", format!("failed to create runtime: {e}").red());
+            return ExitCode::FAILURE;
+        }
+    };
+
+    rt.block_on(run_watch_async(host, port, password, tls, key, interval_ms))
+}
+
+async fn run_watch_async(
+    host: &str,
+    port: u16,
+    password: Option<&str>,
+    tls: Option<&TlsClientConfig>,
+    key: &str,
+    interval_ms: u64,
+) -> ExitCode {
+    let mut conn = match Connection::connect(host, port, tls).await {
+        Ok(c) => c,
+        Err(e) => {
+            eprintln!("{}", format!("could not connect to {host}:{port}: {e}").red());
+            return ExitCode::FAILURE;
+        }
+    };
+
+    if let Some(pw) = password {
+        if let Err(e) = conn.authenticate(pw).await {
+            eprintln!("{}", format!("authentication failed: {e}").red());
+            conn.shutdown().await;
+            return ExitCode::FAILURE;
+        }
+    }
+
+    println!("watching key: {}", key.cyan());
+    println!("interval: {}ms  (ctrl-c to stop)", interval_ms);
+    println!();
+
+    let interval = Duration::from_millis(interval_ms);
+    let mut last_value: Option<String> = None;
+    let mut exit_code = ExitCode::SUCCESS;
+
+    // Pin the shutdown future once so select! can poll it on every iteration
+    // without re-registering the signal handler.
+    let mut shutdown = std::pin::pin!(tokio::signal::ctrl_c());
+
+    loop {
+        tokio::select! {
+            biased;
+
+            _ = &mut shutdown => {
+                println!("\nstopped.");
+                break;
+            }
+
+            _ = tokio::time::sleep(interval) => {
+                match conn.send_command_strs(&["GET", key]).await {
+                    Ok(frame) => {
+                        let formatted = format_response(&frame);
+                        if Some(&formatted) != last_value.as_ref() {
+                            println!("{} {}", timestamp().dimmed(), formatted);
+                            last_value = Some(formatted);
+                        }
+                    }
+                    Err(e) => {
+                        eprintln!("{}", format!("error: {e}").red());
+                        exit_code = ExitCode::FAILURE;
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    conn.shutdown().await;
+    exit_code
+}
+
+/// Returns the current Unix timestamp formatted for display.
+fn timestamp() -> String {
+    let secs = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs();
+    format!("[{secs}]")
+}