Add fmsg watch: stream WebSocket events

README.md

@@ -67,6 +67,7 @@ Programmatic clients use API keys issued by fmsg-webapi.
 | `fmsg whoami` | Print the authenticated fmsg address, API URL, and token expiry |
 | `fmsg list` \| `fmsg ls [--limit N] [--offset N]` | List messages for the authenticated user |
 | `fmsg sent [--limit N] [--offset N]` | List messages authored by the authenticated user |
+| `fmsg watch [--events types] [--once] [--timeout D]` | Stream pushed events (new messages, deliveries) over the server's WebSocket until Ctrl-C |
 | `fmsg get <message-id>` | Retrieve a message by ID, including the short text body for `text/*` messages |
 | `fmsg send <recipient> <file\|text\|->` | Send a message (file path, text, or `-` for stdin) |
 | `fmsg draft create <recipient> <file\|text\|->` | Create a draft message without sending |
@@ -124,6 +125,15 @@ callers:
 - `get-data` without an output file still streams raw body bytes to stdout —
  `--json` does not change it.
 - Errors are unchanged: plain text on stderr, exit code 1.
+- `watch` streams one JSON line per event as it arrives —
+ `{"type":"new_msg","data":{...}}` with `data` in the `list` item shape —
+ preceded by `{"type":"ready"}` once the socket is open (and again after
+ every automatic reconnect, since events may have been missed while it was
+ down: do a `list` catch-up when you see it). Filter with `--events
+ new_msg,delivered,recipients_added`; `--once` exits after the first
+ matching event; `--timeout 30s` stops after that long. Exit code 0 after an
+ event or Ctrl-C, **2** if `--once`/`--timeout` ended before any event, 1 on
+ error.
 
 Note: flags must precede a negative message index (`fmsg --json get -1`), since
 everything after the negative index is treated as positional.
@@ -156,6 +166,11 @@ fmsg --json get 101 | jq .pid
 fmsg sent
 fmsg sent --limit 10 --offset 20
 
+# Wait for pushed events (new messages, delivery confirmations)
+fmsg watch # until Ctrl-C
+fmsg --json watch --events new_msg --once # one JSON line, then exit
+fmsg --json watch --timeout 30s # exit 2 if nothing arrived
+
 # Get a specific message
 fmsg get 101
 

cmd/root.go

@@ -3,6 +3,7 @@ package cmd
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"unicode"
@@ -67,6 +68,9 @@ func Execute() {
 	injectDashDash()
 	if err := rootCmd.Execute(); err != nil {
  fmt.Fprintln(os.Stderr, err)
+ if errors.Is(err, errNoEvent) {
+ os.Exit(exitNoEvent)
+ }
  os.Exit(1)
 	}
 }

cmd/update.go

@@ -1,10 +1,12 @@
 package cmd
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
 	"os"
+	"strconv"
 
 	"github.com/spf13/cobra"
 )
@@ -26,7 +28,9 @@ var updateCmd = &cobra.Command{
  - A text string
  - "-" to read from stdin
 
-Only provided fields are updated; recipients in to are fully replaced.`,
+Only provided fields are updated; recipients in to are fully replaced.
+(The API's PUT replaces the whole draft, so the current draft is fetched
+and merged first — unchanged fields are preserved.)`,
 	Args: cobra.RangeArgs(1, 2),
 	RunE: func(cmd *cobra.Command, args []string) error {
  client, manager := newAuthenticatedClient()
@@ -40,32 +44,56 @@ Only provided fields are updated; recipients in to are fully replaced.`,
  return err
  }
 
+ // The API's PUT replaces the whole draft: fetch the current state and
+ // merge, so fields the caller didn't provide are preserved rather
+ // than wiped.
+ idStr := strconv.FormatInt(msgID, 10)
+ existing, err := client.GetMessage(idStr)
+ if err != nil {
+ return fmt.Errorf("fetching current draft: %w", err)
+ }
+
  msg := map[string]interface{}{
  "from": user,
  "version": 1,
  }
 
+ to := existing.To
  if len(updateTo) > 0 {
- msg["to"] = updateTo
+ to = updateTo
+ }
+ if len(to) > 0 {
+ msg["to"] = to
  }
+
+ topic := existing.Topic
  if cmd.Flags().Changed("topic") {
-msg["topic"] = updateTopic
+ topic = updateTopic
  }
- if cmd.Flags().Changed("type") {
- msg["type"] = updateType
+ if topic != "" {
+ msg["topic"] = topic
  }
+
  if cmd.Flags().Changed("pid") {
  msg["pid"] = updatePID
+ } else if existing.PID != nil {
+ msg["pid"] = *existing.PID
  }
+
+ important := existing.Important
  if cmd.Flags().Changed("important") {
-msg["important"] = updateImportant
+ important = updateImportant
  }
+ msg["important"] = important
+
+ noReply := existing.NoReply
  if cmd.Flags().Changed("no-reply") {
-msg["no_reply"] = updateNoReply
+noReply = updateNoReply
  }
+ msg["no_reply"] = noReply
 
+ var data []byte
  if len(args) == 2 {
- var data []byte
  content := args[1]
  switch content {
  case "-":
@@ -80,11 +108,25 @@ Only provided fields are updated; recipients in to are fully replaced.`,
  data = []byte(content)
  }
  }
-	msg["data"] = string(data)
-msg["size"] = len(data)
- if !cmd.Flags().Changed("type") {
-msg["type"] = "text/plain"
+} else if existing.Size > 0 {
+var buf bytes.Buffer
+ if err := client.DownloadDataToWriter(idStr, &buf); err != nil {
+return fmt.Errorf("fetching current draft body: %w", err)
  }
+ data = buf.Bytes()
+ }
+ msg["data"] = string(data)
+ msg["size"] = len(data)
+
+ typ := existing.Type
+ if cmd.Flags().Changed("type") {
+ typ = updateType
+ }
+ if typ == "" && len(args) == 2 {
+ typ = "text/plain"
+ }
+ if typ != "" {
+ msg["type"] = typ
  }
 
  payload, err := json.Marshal(msg)

cmd/watch.go

@@ -0,0 +1,136 @@
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/markmnl/fmsg-cli/internal/api"
+	"github.com/spf13/cobra"
+)
+
+var (
+	watchEvents []string
+	watchOnce bool
+	watchTimeout time.Duration
+)
+
+// exitNoEvent is the exit code when --once/--timeout ends without an event —
+// distinct from 1 (error) so scripts can tell "nothing arrived" from failure.
+const exitNoEvent = 2
+
+// errNoEvent is returned by watch when it ends without having printed an
+// event; Execute maps it to exitNoEvent.
+var errNoEvent = errors.New("no event received")
+
+var watchCmd = &cobra.Command{
+	Use: "watch",
+	Short: "Stream new-message notifications over the server's WebSocket",
+	Long: `Connect to the fmsg-webapi WebSocket and print each pushed event as it
+arrives, until interrupted (Ctrl-C), --once, or --timeout.
+
+Events pushed by the server: new_msg (a message arrived for you), delivered
+(a message you sent reached a recipient), recipients_added. Filter with
+--events; every event carries the message in the same shape as a "list" item.
+
+With --json each event is one JSON line: {"type":"new_msg","data":{...}}.
+A {"type":"ready"} line is printed once the socket is open — and again after
+every reconnect, since events may have been missed while it was down; do a
+"list" catch-up when you see it. The connection is redialled automatically
+if it drops.
+
+Exit codes: 0 after an event (--once) or when stopped by Ctrl-C/--timeout;
+2 if --once/--timeout ended before any event; 1 on error.`,
+	Args: cobra.NoArgs,
+	SilenceUsage: true,
+	RunE: func(cmd *cobra.Command, args []string) error {
+ client, _ := newAuthenticatedClient()
+
+ ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
+ defer stop()
+ if watchTimeout > 0 {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, watchTimeout)
+ defer cancel()
+ }
+
+ wanted := map[string]bool{}
+ for _, e := range watchEvents {
+ for _, part := range strings.Split(e, ",") {
+ if part = strings.TrimSpace(part); part != "" {
+ wanted[part] = true
+ }
+ }
+ }
+
+ printed := 0
+ opts := api.WatchOptions{
+ Reconnect: true,
+ OnConnect: func() {
+ if jsonOutput {
+ _ = printJSON(map[string]string{"type": "ready"})
+ } else {
+ fmt.Fprintf(os.Stderr, "watching %s (Ctrl-C to stop)\n", client.WatchURL())
+ }
+ },
+ }
+ err := client.Watch(ctx, opts, func(ev api.WatchEvent) error {
+ if len(wanted) > 0 && !wanted[ev.Type] {
+ return nil
+ }
+ if jsonOutput {
+ if err := printJSON(ev); err != nil {
+ return err
+ }
+ } else {
+ printHumanEvent(ev)
+ }
+ printed++
+ if watchOnce {
+ return api.ErrStopWatch
+ }
+ return nil
+ })
+
+ switch {
+ case err == nil:
+ return nil
+ case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
+ // Ctrl-C or --timeout: fine unless the caller wanted an event.
+ if printed == 0 && errors.Is(err, context.DeadlineExceeded) {
+ return errNoEvent
+ }
+ return nil
+ default:
+ return err
+ }
+	},
+}
+
+// printHumanEvent renders one event as a single line.
+func printHumanEvent(ev api.WatchEvent) {
+	item, err := ev.Item()
+	if err != nil {
+ fmt.Printf("%s %s\n", ev.Type, strings.TrimSpace(string(ev.Data)))
+ return
+	}
+	to, _ := json.Marshal(item.To)
+	line := fmt.Sprintf("%-16s ID: %d From: %s To: %s", ev.Type, item.ID, item.From, string(to))
+	if item.Topic != "" {
+ line += fmt.Sprintf(" Topic: %q", item.Topic)
+	}
+	fmt.Println(line)
+}
+
+func init() {
+	watchCmd.Flags().StringSliceVar(&watchEvents, "events", nil, "Only print these event types (comma-separated: new_msg,delivered,recipients_added); default all")
+	watchCmd.Flags().BoolVar(&watchOnce, "once", false, "Exit after the first matching event")
+	watchCmd.Flags().DurationVar(&watchTimeout, "timeout", 0, "Stop after this duration (e.g. 30s, 5m); 0 means run until interrupted")
+	rootCmd.AddCommand(watchCmd)
+}

go.mod

@@ -9,6 +9,7 @@ require (
 )
 
 require (
+	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	golang.org/x/sys v0.35.0 // indirect

go.sum

@@ -1,4 +1,6 @@
 github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=