From f450ba5aa5e599ee9d5b335be3a01f92abda09e7 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Wed, 5 Aug 2026 11:05:12 +1000 Subject: [PATCH 1/2] update: fetch-and-merge before PUT so unchanged fields are preserved The API's PUT replaces the whole draft; update sent only the provided fields, silently wiping recipients, pid, topic, and body. Fetch the current draft first and merge the changed flags over it, making the documented PATCH-style semantics true. Co-Authored-By: Claude Fable 5 --- cmd/update.go | 66 +++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 54 insertions(+), 12 deletions(-) diff --git a/cmd/update.go b/cmd/update.go index 79a13f6..1a6f8bb 100644 --- a/cmd/update.go +++ b/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) From ad8f70ea6fa221d0fbb83c2529d69501435900d7 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Sat, 15 Aug 2026 23:26:50 +1000 Subject: [PATCH 2/2] Add fmsg watch: stream WebSocket events (new_msg, delivered, recipients_added) Wraps GET /fmsg/ws. Token in the Authorization header (401 retried once with a refreshed token), auto-reconnect with backoff, pings answered and the read deadline renewed. --json emits NDJSON with a {"type":"ready"} line after each (re)connect so scripts can catch up with list; --once and --timeout for one-shot use; exit 2 when no event arrived. Co-Authored-By: Claude Fable 5 --- README.md | 15 +++ cmd/root.go | 4 + cmd/watch.go | 136 +++++++++++++++++++++++++++ go.mod | 1 + go.sum | 2 + internal/api/watch.go | 184 ++++++++++++++++++++++++++++++++++++ internal/api/watch_test.go | 187 +++++++++++++++++++++++++++++++++++++ 7 files changed, 529 insertions(+) create mode 100644 cmd/watch.go create mode 100644 internal/api/watch.go create mode 100644 internal/api/watch_test.go diff --git a/README.md b/README.md index 762bc83..ca343fe 100644 --- a/README.md +++ b/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 ` | Retrieve a message by ID, including the short text body for `text/*` messages | | `fmsg send ` | Send a message (file path, text, or `-` for stdin) | | `fmsg draft create ` | 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 diff --git a/cmd/root.go b/cmd/root.go index cba7e65..ef82a93 100644 --- a/cmd/root.go +++ b/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) } } diff --git a/cmd/watch.go b/cmd/watch.go new file mode 100644 index 0000000..71d3975 --- /dev/null +++ b/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) +} diff --git a/go.mod b/go.mod index 996ff62..0696e3f 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index cdd5e4a..a8297b1 100644 --- a/go.sum +++ b/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= diff --git a/internal/api/watch.go b/internal/api/watch.go new file mode 100644 index 0000000..ed54fb7 --- /dev/null +++ b/internal/api/watch.go @@ -0,0 +1,184 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/gorilla/websocket" +) + +// Event type discriminators pushed over GET /fmsg/ws. Unknown types are +// passed through untouched so new server events reach callers unchanged. +const ( + EventNewMsg = "new_msg" + EventDelivered = "delivered" + EventRecipientsAdded = "recipients_added" +) + +// WatchEvent is one frame from the WebSocket: a type discriminator and the +// event-specific body, kept raw so it can be re-emitted byte-for-byte. +type WatchEvent struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` +} + +// Item decodes Data as a message list item (the shape of every event the +// server currently sends). +func (e WatchEvent) Item() (*MessageListItem, error) { + var item MessageListItem + if err := json.Unmarshal(e.Data, &item); err != nil { + return nil, fmt.Errorf("decoding %s event: %w", e.Type, err) + } + return &item, nil +} + +// ErrStopWatch may be returned by a WatchHandler to end Watch cleanly. +var ErrStopWatch = errors.New("stop watch") + +// WatchHandler receives each event. Returning ErrStopWatch ends the watch +// with a nil error; any other error ends it with that error. +type WatchHandler func(WatchEvent) error + +// WatchOptions tunes Watch. +type WatchOptions struct { + // OnConnect is called after every successful handshake (including + // reconnects), before any event from that connection is delivered. + OnConnect func() + // Reconnect re-dials with backoff when the connection drops instead of + // returning the read error. Handshake failures other than an expired + // token (401, retried once with a fresh token) are never retried. + Reconnect bool + // Dialer overrides the WebSocket dialer (tests). + Dialer *websocket.Dialer +} + +// readTimeout bounds silence on the socket. The server pings every 45s, so a +// healthy connection always produces a frame well inside this window. +const readTimeout = 90 * time.Second + +// Watch connects to GET /fmsg/ws and calls fn for every pushed event until +// ctx is cancelled, fn returns an error, or (with Reconnect off) the +// connection drops. +func (c *Client) Watch(ctx context.Context, opts WatchOptions, fn WatchHandler) error { + backoff := time.Second + for { + conn, err := c.dialWatch(ctx, opts.Dialer) + if err != nil { + return err + } + if opts.OnConnect != nil { + opts.OnConnect() + } + backoff = time.Second + err = readEvents(ctx, conn, fn) + conn.Close() + if ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, ErrStopWatch) { + return nil + } + var readErr *watchReadError + if !opts.Reconnect || !errors.As(err, &readErr) { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + if backoff < 30*time.Second { + backoff *= 2 + } + } +} + +// watchReadError marks a connection-level failure (as opposed to a handler +// error) so Watch knows a reconnect is appropriate. +type watchReadError struct{ err error } + +func (e *watchReadError) Error() string { return "websocket: " + e.err.Error() } +func (e *watchReadError) Unwrap() error { return e.err } + +// WatchURL returns the WebSocket endpoint derived from BaseURL. +func (c *Client) WatchURL() string { + u := c.BaseURL + "/fmsg/ws" + switch { + case strings.HasPrefix(u, "https://"): + return "wss://" + strings.TrimPrefix(u, "https://") + case strings.HasPrefix(u, "http://"): + return "ws://" + strings.TrimPrefix(u, "http://") + } + return u +} + +// dialWatch performs the handshake with the bearer token in the +// Authorization header (never the query string, which lands in logs). A 401 +// is retried once with a force-refreshed token, mirroring Client.do. +func (c *Client) dialWatch(ctx context.Context, dialer *websocket.Dialer) (*websocket.Conn, error) { + if c.Auth == nil { + return nil, fmt.Errorf("missing token provider") + } + if dialer == nil { + dialer = websocket.DefaultDialer + } + dial := func(forceRefresh bool) (*websocket.Conn, *http.Response, error) { + token, err := c.Auth.AccessToken(ctx, forceRefresh) + if err != nil { + return nil, nil, err + } + h := http.Header{} + h.Set("Authorization", "Bearer "+token) + return dialer.DialContext(ctx, c.WatchURL(), h) + } + conn, resp, err := dial(false) + if err != nil && resp != nil && resp.StatusCode == http.StatusUnauthorized { + conn, resp, err = dial(true) + } + if err != nil { + if resp != nil { + return nil, &apiError{StatusCode: resp.StatusCode, Body: http.StatusText(resp.StatusCode)} + } + return nil, fmt.Errorf("network error: %w", err) + } + return conn, nil +} + +func readEvents(ctx context.Context, conn *websocket.Conn, fn WatchHandler) error { + // Close the socket when ctx ends so a blocked ReadMessage returns. + stop := context.AfterFunc(ctx, func() { conn.Close() }) + defer stop() + + conn.SetReadDeadline(time.Now().Add(readTimeout)) + conn.SetPongHandler(func(string) error { return conn.SetReadDeadline(time.Now().Add(readTimeout)) }) + // gorilla's default ping handler answers with a pong; it also needs the + // deadline pushed, since a ping is proof the server is alive. + conn.SetPingHandler(func(data string) error { + conn.SetReadDeadline(time.Now().Add(readTimeout)) + err := conn.WriteControl(websocket.PongMessage, []byte(data), time.Now().Add(10*time.Second)) + if err == websocket.ErrCloseSent { + return nil + } + return err + }) + + for { + _, payload, err := conn.ReadMessage() + if err != nil { + return &watchReadError{err} + } + conn.SetReadDeadline(time.Now().Add(readTimeout)) + var ev WatchEvent + if err := json.Unmarshal(payload, &ev); err != nil { + return &watchReadError{fmt.Errorf("decoding event: %w", err)} + } + if err := fn(ev); err != nil { + return err + } + } +} diff --git a/internal/api/watch_test.go b/internal/api/watch_test.go new file mode 100644 index 0000000..61d0090 --- /dev/null +++ b/internal/api/watch_test.go @@ -0,0 +1,187 @@ +package api + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +// refreshingProvider hands out "stale" first, then "fresh" once refreshed. +type refreshingProvider struct{ refreshed atomic.Bool } + +func (p *refreshingProvider) AccessToken(ctx context.Context, force bool) (string, error) { + if force { + p.refreshed.Store(true) + } + if p.refreshed.Load() { + return "fresh", nil + } + return "stale", nil +} + +// wsServer serves /fmsg/ws, accepting only Bearer "fresh", and runs serve on +// each upgraded connection. +func wsServer(t *testing.T, serve func(*websocket.Conn)) *httptest.Server { + t.Helper() + up := websocket.Upgrader{} + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/fmsg/ws" { + http.NotFound(w, r) + return + } + if r.URL.Query().Get("access_token") != "" { + t.Errorf("token must travel in the header, not the query string") + } + if r.Header.Get("Authorization") != "Bearer fresh" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + conn, err := up.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + serve(conn) + })) +} + +func TestWatchURL(t *testing.T) { + for in, want := range map[string]string{ + "http://127.0.0.1:8000": "ws://127.0.0.1:8000/fmsg/ws", + "https://api.example.com/": "wss://api.example.com/fmsg/ws", + } { + if got := New(in, StaticTokenProvider("x")).WatchURL(); got != want { + t.Errorf("WatchURL(%q) = %q, want %q", in, got, want) + } + } +} + +func TestWatchDeliversEventsRefreshesTokenAndStops(t *testing.T) { + srv := wsServer(t, func(conn *websocket.Conn) { + conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"new_msg","data":{"id":7,"from":"@a@x","to":["@b@y"]}}`)) + conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"delivered","data":{"id":8}}`)) + // Keep the connection open until the client goes away. + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + }) + defer srv.Close() + + prov := &refreshingProvider{} + c := New(srv.URL, prov) + var got []WatchEvent + connects := 0 + err := c.Watch(context.Background(), WatchOptions{OnConnect: func() { connects++ }}, func(ev WatchEvent) error { + got = append(got, ev) + if len(got) == 2 { + return ErrStopWatch + } + return nil + }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + if !prov.refreshed.Load() { + t.Errorf("expected a forced token refresh after 401") + } + if connects != 1 { + t.Errorf("OnConnect called %d times, want 1", connects) + } + if len(got) != 2 || got[0].Type != "new_msg" || got[1].Type != "delivered" { + t.Fatalf("events = %+v", got) + } + item, err := got[0].Item() + if err != nil || item.ID != 7 || item.From != "@a@x" { + t.Errorf("Item() = %+v, %v", item, err) + } +} + +func TestWatchHandlerErrorPropagates(t *testing.T) { + srv := wsServer(t, func(conn *websocket.Conn) { + conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"new_msg","data":{"id":1}}`)) + conn.ReadMessage() + }) + defer srv.Close() + boom := errors.New("boom") + err := New(srv.URL, StaticTokenProvider("fresh")).Watch(context.Background(), WatchOptions{}, func(WatchEvent) error { return boom }) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } +} + +func TestWatchReconnectsAndReportsReady(t *testing.T) { + var conns atomic.Int32 + srv := wsServer(t, func(conn *websocket.Conn) { + n := conns.Add(1) + if n == 1 { + return // drop the first connection immediately + } + conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"new_msg","data":{"id":2}}`)) + conn.ReadMessage() + }) + defer srv.Close() + + ready := 0 + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + err := New(srv.URL, StaticTokenProvider("fresh")).Watch(ctx, WatchOptions{ + Reconnect: true, + OnConnect: func() { ready++ }, + }, func(ev WatchEvent) error { return ErrStopWatch }) + if err != nil { + t.Fatalf("Watch: %v", err) + } + if ready != 2 || conns.Load() != 2 { + t.Errorf("ready=%d conns=%d, want 2/2", ready, conns.Load()) + } +} + +func TestWatchWithoutReconnectReturnsDropError(t *testing.T) { + srv := wsServer(t, func(conn *websocket.Conn) {}) + defer srv.Close() + err := New(srv.URL, StaticTokenProvider("fresh")).Watch(context.Background(), WatchOptions{}, func(WatchEvent) error { return nil }) + if err == nil || !strings.Contains(err.Error(), "websocket") { + t.Fatalf("err = %v, want websocket read error", err) + } +} + +func TestWatchContextCancel(t *testing.T) { + srv := wsServer(t, func(conn *websocket.Conn) { conn.ReadMessage() }) + defer srv.Close() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- New(srv.URL, StaticTokenProvider("fresh")).Watch(ctx, WatchOptions{Reconnect: true}, func(WatchEvent) error { return nil }) + }() + time.Sleep(100 * time.Millisecond) + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Watch did not return after cancel") + } +} + +func TestWatchHandshakeFailureIsAPIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "forbidden", http.StatusForbidden) + })) + defer srv.Close() + err := New(srv.URL, StaticTokenProvider("fresh")).Watch(context.Background(), WatchOptions{Reconnect: true}, func(WatchEvent) error { return nil }) + var ae *apiError + if !errors.As(err, &ae) || ae.StatusCode != http.StatusForbidden { + t.Fatalf("err = %v, want apiError 403", err) + } +}