diff --git a/.surface b/.surface index 0f59eb51..19194ec4 100644 --- a/.surface +++ b/.surface @@ -170,3 +170,11 @@ hey tui hey unseen hey upgrade hey version +hey watch +hey watch --box +hey watch --events +hey watch --exit-on-first +hey watch --run-async +hey watch --run-sync +hey watch --since +hey watch --timeout diff --git a/AGENTS.md b/AGENTS.md index a2dab598..e5b5607d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,6 +94,23 @@ The TUI renders inline images using the Kitty graphics protocol's Unicode Placeh This works in Kitty and Ghostty. Other terminals show the text content normally (placeholders are invisible). +### Watching for changes over Action Cable + +`hey watch` is told when a box changed instead of polling for it. +`internal/cable` dials HEY's cable server with [actioncable-go](github.com/basecamp/actioncable-go), +authorizing the upgrade request with the same credentials the SDK sends on an API request +(`HEY_CABLE_URL` overrides the endpoint). `internal/cmd/watch.go` subscribes to +haystack's `Postings::ChangesChannel`, which broadcasts only `{change, account_id, box_id, +box_kind, posting_ids, at}` — a doorbell, not the change itself. + +The change is then read through `Postings().AllChanges`, the same incremental sync feed the +mail clients use, starting from the cursor in the box's `posting_changes_url`. That is what +makes a reconnect safe: the cursor, not the notification, is the source of truth, so a +missed broadcast costs nothing, and a 409 means catch up in full instead. A read that +fails leaves the cursor where it was and is retried on a doubling backoff, so a change +isn't lost with the notification that announced it, and a subscription that closes without +the watch being interrupted is an error rather than a quiet exit. + ### API documentation If you are unsure what the API endpoints are, what they expect or what they respond to you can read through the server implementation to understand how the API works. diff --git a/API-COVERAGE.md b/API-COVERAGE.md index d921606d..ee2b94d1 100644 --- a/API-COVERAGE.md +++ b/API-COVERAGE.md @@ -65,3 +65,5 @@ The remaining HTML-reading gaps use the SDK's authenticated HTML helper and are | `/calendar/todos/{id}/completions.json` | POST | SDK `CalendarTodos().Complete` | `hey todo complete ` | covered | | `/calendar/todos/{id}/completions.json` | DELETE | SDK `CalendarTodos().Uncomplete` | `hey todo uncomplete ` | covered | | `/calendar/todos/{id}.json` | DELETE | SDK `CalendarTodos().Delete` | `hey todo delete ` | covered | +| `/boxes/{id}/postings/changes.json` | GET | SDK `Postings().AllChanges` | `hey watch` | covered | +| `/cable` (`Postings::ChangesChannel`) | WS | `internal/cable` + actioncable-go | `hey watch` | covered | diff --git a/README.md b/README.md index 4b79cdb5..bea9df02 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,33 @@ Contact updates preserve omitted name, email, and alias fields. Supplying `--ali Organization actions take the `id` values returned by `hey box --json`, `hey label --json`, or `hey search --json`. Label IDs come from `hey labels`; `hey label` returns `next_page` and `total_count`, accepts `--page ` for continuation, and supports `--all` for complete traversal. HEY creates a label while adding it to at least one thread, so `hey label create` requires thread item IDs. Move destinations are Imbox, The Feed, Set Aside, Reply Later, or Paper Trail. Bubble Up requires a scheduled date and is not available through `hey move`. Trashing a shared thread removes your access instead of deleting it for everyone. Ignored threads remain in their box and can be restored with `hey stop-ignoring`. +### Watching for changes + +```bash +hey watch # follow every box, a line of JSON per change +hey watch --box imbox --events added # only new postings in the Imbox +hey watch --box imbox --exit-on-first # block until something lands, then exit +hey watch --since 2026-08-18T09:00:00Z # catch up from a time first, then follow +hey watch --run-async 'notify-send "New mail in $HEY_BOX_KIND"' +hey watch --run-sync ./triage.sh # one at a time, waiting for each +``` + +Runs until interrupted, printing changes as they happen, one line each: + +```json +{"change":"added","at":"2026-08-18T09:14:22.031Z","box":{"id":24088,"kind":"imbox","name":"Imbox"},"posting_id":98765,"thread_id":54321,"posting":{}} +``` + +A change can drive a command instead of being printed, and there's a choice to make +between two behaviours — pass one or the other, not both. `--run-async` spawns the +command per change and moves on, so a slow one never holds up the watch and two can +overlap. `--run-sync` waits for each and runs them in order, so they never overlap and a +slow one delays the next. + +Both hand the JSON to the command on its stdin, and the same fields as `HEY_CHANGE`, +`HEY_AT`, `HEY_BOX_ID`, `HEY_BOX_KIND`, `HEY_BOX_NAME`, `HEY_POSTING_ID` and +`HEY_THREAD_ID`. Both also take over stdout, so the JSON isn't printed as well. + ### Calendars ```bash diff --git a/go.mod b/go.mod index 10bcbb20..5682476e 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( charm.land/bubbles/v2 v2.1.1 charm.land/bubbletea/v2 v2.0.8 charm.land/lipgloss/v2 v2.0.6 + github.com/basecamp/actioncable-go v0.0.0-20260819125529-39dd29b8e4d1 github.com/basecamp/hey-sdk/go v0.6.1 github.com/charmbracelet/x/ansi v0.11.8 github.com/gofrs/flock v0.13.0 diff --git a/go.sum b/go.sum index 988b8dd7..5ad32a90 100644 --- a/go.sum +++ b/go.sum @@ -77,6 +77,8 @@ github.com/aws/smithy-go v1.26.0 h1:9ouqbi+NyKP7fV3Te7UElCwdAb6Y8uk7LGwPE5tVe/s= github.com/aws/smithy-go v1.26.0/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/aymanbagabas/go-udiff v0.4.1 h1:OEIrQ8maEeDBXQDoGCbbTTXYJMYRCRO1fnodZ12Gv5o= github.com/aymanbagabas/go-udiff v0.4.1/go.mod h1:0L9PGwj20lrtmEMeyw4WKJ/TMyDtvAoK9bf2u/mNo3w= +github.com/basecamp/actioncable-go v0.0.0-20260819125529-39dd29b8e4d1 h1:aNCaFvx7FosjImt1A3TM3aWizOgWGHtKUoQ4RTa6Mx8= +github.com/basecamp/actioncable-go v0.0.0-20260819125529-39dd29b8e4d1/go.mod h1:9+DEydJMniIKraEsd4fDJpFEnqlLUJ6XhAswxRBaITk= github.com/basecamp/hey-sdk/go v0.6.1 h1:NlruAUq1GOk+VCkc1HyxAsvLCbNf54n0kZ6kNfhZkyc= github.com/basecamp/hey-sdk/go v0.6.1/go.mod h1:k6sO2XhMkU3UY8lD2ozp0735Ic3q8xoMQt7YUT3TlYk= github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ= diff --git a/internal/cable/cable.go b/internal/cable/cable.go new file mode 100644 index 00000000..79ca660d --- /dev/null +++ b/internal/cable/cable.go @@ -0,0 +1,77 @@ +// Package cable connects to HEY's Action Cable server, so commands can be told +// when something changed instead of polling for it. +package cable + +import ( + "context" + "fmt" + "net/http" + "net/url" + "os" + "strings" + + "github.com/basecamp/actioncable-go" + + "github.com/basecamp/hey-cli/internal/auth" +) + +// Dial connects to the cable server for a HEY base URL, authorizing the upgrade +// request with the same credentials the SDK sends on an API request. +func Dial(ctx context.Context, baseURL string, authMgr *auth.Manager, options ...actioncable.Option) (*actioncable.Client, error) { + cableURL, err := URL(baseURL) + if err != nil { + return nil, err + } + + header, err := authHeader(ctx, baseURL, authMgr) + if err != nil { + return nil, err + } + + client := actioncable.New(cableURL, append([]actioncable.Option{actioncable.WithHeader(header)}, options...)...) + if err := client.Connect(ctx); err != nil { + return nil, err + } + + return client, nil +} + +// URL is the cable endpoint for a base URL: https://app.hey.com becomes +// wss://app.hey.com/cable. HEY_CABLE_URL overrides it outright. +func URL(baseURL string) (string, error) { + if override := os.Getenv("HEY_CABLE_URL"); override != "" { + return override, nil + } + + parsed, err := url.Parse(strings.TrimSuffix(baseURL, "/")) + if err != nil { + return "", fmt.Errorf("could not read base URL %q: %w", baseURL, err) + } + + switch parsed.Scheme { + case "https": + parsed.Scheme = "wss" + case "http": + parsed.Scheme = "ws" + default: + return "", fmt.Errorf("base URL %q is neither http nor https", baseURL) + } + + parsed.Path = strings.TrimSuffix(parsed.Path, "/") + "/cable" + parsed.RawQuery = "" + + return parsed.String(), nil +} + +func authHeader(ctx context.Context, baseURL string, authMgr *auth.Manager) (http.Header, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL, nil) + if err != nil { + return nil, err + } + + if err := authMgr.AuthenticateRequest(ctx, request); err != nil { + return nil, err + } + + return request.Header, nil +} diff --git a/internal/cable/cable_test.go b/internal/cable/cable_test.go new file mode 100644 index 00000000..5b35f42a --- /dev/null +++ b/internal/cable/cable_test.go @@ -0,0 +1,42 @@ +package cable + +import "testing" + +func TestURL(t *testing.T) { + cases := []struct { + baseURL string + want string + }{ + {"https://app.hey.com", "wss://app.hey.com/cable"}, + {"https://app.hey.com/", "wss://app.hey.com/cable"}, + {"http://app.hey.localhost:3003", "ws://app.hey.localhost:3003/cable"}, + } + + for _, c := range cases { + got, err := URL(c.baseURL) + if err != nil { + t.Fatalf("URL(%q) failed: %v", c.baseURL, err) + } + if got != c.want { + t.Errorf("URL(%q) = %q, want %q", c.baseURL, got, c.want) + } + } +} + +func TestURLOverride(t *testing.T) { + t.Setenv("HEY_CABLE_URL", "ws://cable.example.com/cable") + + got, err := URL("https://app.hey.com") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != "ws://cable.example.com/cable" { + t.Errorf("URL = %q, want the HEY_CABLE_URL override", got) + } +} + +func TestURLRejectsOtherSchemes(t *testing.T) { + if _, err := URL("ftp://app.hey.com"); err == nil { + t.Fatal("expected an error for a base URL that isn't http or https") + } +} diff --git a/internal/cmd/help.go b/internal/cmd/help.go index cba2245c..2f2c908a 100644 --- a/internal/cmd/help.go +++ b/internal/cmd/help.go @@ -17,7 +17,7 @@ var curatedCategories = []struct { }{ { heading: "EMAIL", - names: []string{"boxes", "box", "labels", "label", "search", "contacts", "threads", "attachments", "compose", "reply", "bulk-reply", "forward", "drafts", "seen", "unseen", "move", "trash", "spam", "ignore", "stop-ignoring"}, + names: []string{"boxes", "box", "labels", "label", "search", "contacts", "threads", "attachments", "compose", "reply", "bulk-reply", "forward", "drafts", "seen", "unseen", "move", "trash", "spam", "ignore", "stop-ignoring", "watch"}, }, { heading: "CALENDAR & TASKS", diff --git a/internal/cmd/help_test.go b/internal/cmd/help_test.go index 4ff5ff54..fa64fea9 100644 --- a/internal/cmd/help_test.go +++ b/internal/cmd/help_test.go @@ -30,7 +30,7 @@ func TestCuratedCommandHelpUsesUserFacingLanguage(t *testing.T) { func TestEmailCommandHelpKeepsPostingAsAnInternalTerm(t *testing.T) { root := newRootCmd() - for _, name := range []string{"boxes", "box", "labels", "label", "search", "seen", "unseen", "move", "trash", "spam", "ignore", "stop-ignoring"} { + for _, name := range []string{"boxes", "box", "labels", "label", "search", "seen", "unseen", "move", "trash", "spam", "ignore", "stop-ignoring", "watch"} { t.Run(name, func(t *testing.T) { command, _, err := root.Find([]string{name}) if err != nil { @@ -112,6 +112,7 @@ EMAIL spam Mark email threads as spam ignore Ignore email threads stop-ignoring Stop ignoring email threads + watch Follow email threads as they change CALENDAR & TASKS calendars List calendars diff --git a/internal/cmd/root.go b/internal/cmd/root.go index d61ef2f7..10720122 100644 --- a/internal/cmd/root.go +++ b/internal/cmd/root.go @@ -172,6 +172,7 @@ func newRootCmd() *cobra.Command { root.AddCommand(newHabitCommand().cmd) root.AddCommand(newTimetrackCommand().cmd) root.AddCommand(newJournalCommand().cmd) + root.AddCommand(newWatchCommand().cmd) root.AddCommand(newSeenCommand().cmd) root.AddCommand(newUnseenCommand().cmd) root.AddCommand(newMoveCommand().cmd) diff --git a/internal/cmd/watch.go b/internal/cmd/watch.go new file mode 100644 index 00000000..739999ee --- /dev/null +++ b/internal/cmd/watch.go @@ -0,0 +1,633 @@ +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "maps" + "os" + "os/exec" + "os/signal" + "runtime" + "slices" + "strconv" + "strings" + "sync/atomic" + "syscall" + "time" + + actioncable "github.com/basecamp/actioncable-go" + "github.com/spf13/cobra" + + "github.com/basecamp/hey-sdk/go/pkg/generated" + hey "github.com/basecamp/hey-sdk/go/pkg/hey" + + "github.com/basecamp/hey-cli/internal/cable" + "github.com/basecamp/hey-cli/internal/output" +) + +const changesChannel = "Postings::ChangesChannel" + +// A failed read of the changes feed is retried on its own, doubling the wait each time so +// a server that stays down isn't hammered, and asyncScriptLimit caps how many --run-async +// commands run at once so a catch-up of thousands of changes can't fork a process for each. +const ( + firstWatchRetry = 2 * time.Second + longestWatchRetry = 2 * time.Minute + asyncScriptLimit = 16 +) + +var watchableChanges = []string{"added", "updated", "deleted"} + +type watchCommand struct { + cmd *cobra.Command + boxes []string + events []string + since string + asyncScript string + syncScript string + exitOnFirst bool + timeout time.Duration +} + +func newWatchCommand() *watchCommand { + watchCommand := &watchCommand{} + watchCommand.cmd = &cobra.Command{ + Use: "watch", + Short: "Follow email threads as they change", + Long: `Print email threads as they change, one JSON object per line. Runs until interrupted. + +Changes can drive a command instead of being printed, and that is a choice between two +behaviours: --run-async spawns the command per change and moves on, so a slow one never +holds up the watch and two can overlap; --run-sync waits for each and runs them in order. +Pass one or the other.`, + Annotations: map[string]string{ + "agent_notes": "Long-running. Writes one JSON object per changed thread to stdout (NDJSON), not the usual envelope. Use --exit-on-first to block until one change lands and then exit.", + }, + Example: ` hey watch + hey watch --box imbox --events added + hey watch --box imbox --exit-on-first + hey watch --run-async 'notify-send "New mail in $HEY_BOX_KIND"' + hey watch --run-sync ./triage.sh + hey watch --since 2026-08-18T09:00:00Z`, + RunE: watchCommand.run, + Args: cobra.NoArgs, + } + + flags := watchCommand.cmd.Flags() + flags.StringArrayVar(&watchCommand.boxes, "box", nil, "Box to watch by name or ID (repeatable, defaults to all)") + flags.StringSliceVar(&watchCommand.events, "events", watchableChanges, "Changes to report: added, updated, deleted") + flags.StringVar(&watchCommand.since, "since", "", "Report changes since this time first (RFC 3339 or YYYY-MM-DD)") + flags.StringVar(&watchCommand.asyncScript, "run-async", "", "Shell command to spawn per change, without waiting for it") + flags.StringVar(&watchCommand.syncScript, "run-sync", "", "Shell command to run per change, one at a time, waiting for each") + flags.BoolVar(&watchCommand.exitOnFirst, "exit-on-first", false, "Exit after the first change") + flags.DurationVar(&watchCommand.timeout, "timeout", 0, "Give up waiting after this long (for example 30m)") + + return watchCommand +} + +func (c *watchCommand) run(cmd *cobra.Command, args []string) error { + if err := requireAuth(); err != nil { + return err + } + + changes, err := c.watchedChanges() + if err != nil { + return err + } + if c.asyncScript != "" && c.syncScript != "" { + return output.ErrUsage("pass either --run-async or --run-sync, not both") + } + + ctx, stopListeningForSignals := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer stopListeningForSignals() + + if c.timeout > 0 { + timed, giveUp := context.WithTimeout(ctx, c.timeout) + defer giveUp() + ctx = timed + } + + boxes, err := c.watchedBoxes(ctx) + if err != nil { + return err + } + + watch := &postingsWatch{ + boxes: boxes, + changes: changes, + asyncScript: c.asyncScript, + syncScript: c.syncScript, + exitOnFirst: c.exitOnFirst, + out: cmd.OutOrStdout(), + errOut: cmd.ErrOrStderr(), + styled: writer.IsStyled(), + catchUp: make(chan struct{}, 1), + unread: map[int64]bool{}, + running: make(chan struct{}, asyncScriptLimit), + } + + client, err := cable.Dial(ctx, cfg.BaseURL, authMgr) + if err != nil { + return watchDialError(err) + } + defer func() { _ = client.Close() }() + + subscription, err := client.Subscribe(ctx, actioncable.Identifier{Channel: changesChannel}, + actioncable.OnConnected(func(reconnected bool) { + if reconnected { + watch.askForCatchUp() + } + }), + actioncable.OnRejected(func() { watch.rejected.Store(true) })) + if err != nil { + return output.ErrAPI(0, fmt.Sprintf("could not subscribe to posting changes: %v", err)) + } + + if err := watch.listen(ctx, subscription); err != nil { + return err + } + + // A synchronous script's verdict is the command's verdict when we only waited + // for the one change, and there's no other way to answer with its exit code. + if c.exitOnFirst && watch.lastScriptExit != 0 { + os.Exit(watch.lastScriptExit) + } + + return nil +} + +// watchDialError tells the two ways a dial fails apart: the server turned the +// credentials down, or it couldn't be reached at all. +func watchDialError(err error) error { + var disconnect *actioncable.DisconnectError + if errors.As(err, &disconnect) && disconnect.Reason == actioncable.ReasonUnauthorized { + return output.ErrAuth("HEY's cable server turned these credentials down — run `hey auth login` again, or log in with `hey auth login --cookie` if the server doesn't take access tokens on a websocket yet") + } + + return output.ErrNetwork(fmt.Errorf("could not connect to HEY's cable server: %w", err)) +} + +func (c *watchCommand) watchedChanges() (map[string]bool, error) { + changes := map[string]bool{} + for _, event := range c.events { + event = strings.ToLower(strings.TrimSpace(event)) + if !slices.Contains(watchableChanges, event) { + return nil, output.ErrUsage(fmt.Sprintf("unknown event %q — pass any of %s", event, strings.Join(watchableChanges, ", "))) + } + changes[event] = true + } + + if len(changes) == 0 { + return nil, output.ErrUsage("--events needs at least one of " + strings.Join(watchableChanges, ", ")) + } + + return changes, nil +} + +func (c *watchCommand) watchedBoxes(ctx context.Context) (map[int64]*watchedBox, error) { + listed, err := sdk.Boxes().List(ctx) + if err != nil { + return nil, convertSDKError(err) + } + if listed == nil { + return nil, output.ErrAPI(0, "could not list boxes") + } + + watched := map[int64]*watchedBox{} + for _, box := range *listed { + if !c.watching(box) { + continue + } + + cursor, err := watchCursor(box.PostingChangesUrl, c.since) + if err != nil { + return nil, err + } + if cursor.Since == "" { + continue + } + + watched[box.Id] = &watchedBox{id: box.Id, kind: box.Kind, name: box.Name, cursor: cursor} + } + + if len(watched) == 0 { + return nil, output.ErrNotFound("box", strings.Join(c.boxes, ", ")) + } + + return watched, nil +} + +func (c *watchCommand) watching(box generated.Box) bool { + if len(c.boxes) == 0 { + return true + } + + return slices.ContainsFunc(c.boxes, func(wanted string) bool { + return strings.EqualFold(wanted, box.Kind) || + strings.EqualFold(wanted, box.Name) || + wanted == strconv.FormatInt(box.Id, 10) + }) +} + +// watchCursor is where a box's changes feed should be read from. The server bakes its own +// clock into the box's changes URL, so that's the cursor unless --since moves it. +func watchCursor(changesURL, since string) (hey.PostingChangesCursor, error) { + if changesURL == "" { + return hey.PostingChangesCursor{}, nil + } + + cursor, err := hey.PostingChangesCursorFrom(changesURL) + if err != nil { + return hey.PostingChangesCursor{}, err + } + if since == "" { + return cursor, nil + } + + at, err := parseWatchSince(since) + if err != nil { + return hey.PostingChangesCursor{}, err + } + cursor.Since = at.UTC().Format("2006-01-02T15:04:05.000Z") + + return cursor, nil +} + +func parseWatchSince(since string) (time.Time, error) { + if at, err := time.Parse(time.RFC3339, since); err == nil { + return at, nil + } + if at, err := time.Parse("2006-01-02", since); err == nil { + return at, nil + } + return time.Time{}, output.ErrUsage(fmt.Sprintf("could not read --since %q — pass an RFC 3339 time or YYYY-MM-DD", since)) +} + +type watchedBox struct { + id int64 + kind string + name string + cursor hey.PostingChangesCursor +} + +// watchEvent is one changed posting, as a line of NDJSON or as a script's stdin. +type watchEvent struct { + Change string `json:"change"` + At string `json:"at"` + Box watchEventBox `json:"box"` + PostingID int64 `json:"posting_id"` + ThreadID int64 `json:"thread_id,omitempty"` + Posting *generated.Posting `json:"posting,omitempty"` +} + +type watchEventBox struct { + ID int64 `json:"id"` + Kind string `json:"kind"` + Name string `json:"name"` +} + +// postingsWatch holds what a run of the command follows: the boxes and how far each +// one has been read, and what to do with a change once it arrives. +type postingsWatch struct { + boxes map[int64]*watchedBox + changes map[string]bool + asyncScript string + syncScript string + exitOnFirst bool + out io.Writer + errOut io.Writer + styled bool + catchUp chan struct{} + rejected atomic.Bool + unread map[int64]bool + backoff time.Duration + retry <-chan time.Time + running chan struct{} + reported int + lastScriptExit int +} + +func (w *postingsWatch) listen(ctx context.Context, subscription *actioncable.Subscription) error { + if err := w.readEveryBox(ctx); err != nil { + return err + } + + for !w.finished() { + select { + case <-ctx.Done(): + // An interrupt or a --timeout is how a watch is meant to end. + return nil + case <-w.catchUp: + if err := w.readEveryBox(ctx); err != nil { + return err + } + case <-w.retry: + w.retry = nil + if err := w.readUnreadBoxes(ctx); err != nil { + return err + } + case message, open := <-subscription.Messages(): + if !open { + return w.closedError(ctx) + } + if err := w.read(ctx, message); err != nil { + return err + } + } + } + + return nil +} + +// closedError tells the two ways the subscription's messages dry up apart: the watch was +// interrupted or timed out, which is how it's meant to end, or the connection went away +// for good and there is nothing left listening — which a watch left running unattended +// has to hear about rather than exiting quietly. +func (w *postingsWatch) closedError(ctx context.Context) error { + switch { + case ctx.Err() != nil: + return nil //nolint:nilerr // an interrupt or a --timeout is how a watch is meant to end + case w.rejected.Load(): + return output.ErrAuth("HEY's cable server turned this subscription down — run `hey auth login` again, or log in with `hey auth login --cookie` if the server doesn't take access tokens on a websocket yet") + default: + return output.ErrNetwork(errors.New("HEY's cable server hung up for good — nothing is watching for changes any more")) + } +} + +// askForCatchUp is called from the cable client's own goroutine, so it drops the ask +// when one is already waiting rather than blocking the connection. +func (w *postingsWatch) askForCatchUp() { + select { + case w.catchUp <- struct{}{}: + default: + } +} + +func (w *postingsWatch) readEveryBox(ctx context.Context) error { + for _, id := range slices.Sorted(maps.Keys(w.boxes)) { + if err := w.readBox(ctx, w.boxes[id]); err != nil { + return err + } + } + + return nil +} + +// readUnreadBoxes retries the boxes whose last read failed. Their cursors haven't moved, +// so the changes they missed are still ahead of them. +func (w *postingsWatch) readUnreadBoxes(ctx context.Context) error { + for _, id := range slices.Sorted(maps.Keys(w.unread)) { + if err := w.readBox(ctx, w.boxes[id]); err != nil { + return err + } + } + + return nil +} + +func (w *postingsWatch) read(ctx context.Context, message actioncable.Message) error { + var notification struct { + BoxID int64 `json:"box_id"` + } + if err := message.Unmarshal(¬ification); err != nil { + fmt.Fprintf(w.errOut, "warning: could not read a change notification: %v\n", err) + return nil + } + + if box, watching := w.boxes[notification.BoxID]; watching { + return w.readBox(ctx, box) + } + + return nil +} + +func (w *postingsWatch) readBox(ctx context.Context, box *watchedBox) error { + changes, err := sdk.Postings().AllChanges(ctx, box.id, box.cursor) + if err != nil { //nolint:nilerr // a watch reports a failed read and keeps listening + if ctx.Err() == nil { + fmt.Fprintf(w.errOut, "warning: could not read changes in %s: %v\n", box.name, err) + w.readAgainLater(box) + } + return nil + } + w.wasRead(box) + + if changes.FullSyncRequired { + fmt.Fprintf(w.errOut, "notice: too much changed in %s to follow one change at a time — skipping ahead, read the box with `hey box %s`\n", box.name, box.kind) + return w.skipAhead(ctx, box) + } + + if changes.NextCursor != nil { + box.cursor = *changes.NextCursor + } + + for _, posting := range changes.Added { + w.report(ctx, watchEvent{Change: "added", At: watchTime(posting.CreatedAt), PostingID: posting.Id}, box, &posting) + } + for _, posting := range changes.Updated { + w.report(ctx, watchEvent{Change: "updated", At: watchTime(posting.UpdatedAt), PostingID: posting.Id}, box, &posting) + } + for _, posting := range changes.Deleted { + w.report(ctx, watchEvent{Change: "deleted", At: watchTime(posting.DeletedAt), PostingID: posting.Id}, box, nil) + } + + return nil +} + +// readAgainLater keeps a box that couldn't be read on the list, and arms the retry that +// comes back to it. Without it a failed read consumes the notification that prompted it, +// and the change stays invisible until the next email happens along. +func (w *postingsWatch) readAgainLater(box *watchedBox) { + w.unread[box.id] = true + + if w.retry == nil { + w.backoff = min(max(2*w.backoff, firstWatchRetry), longestWatchRetry) + w.retry = time.After(w.backoff) + } +} + +func (w *postingsWatch) wasRead(box *watchedBox) { + delete(w.unread, box.id) + + if len(w.unread) == 0 { + w.backoff = 0 + } +} + +// skipAhead moves a box's cursor to the server's current one, which is the only way +// back once a box has changed more than an increment can carry. +func (w *postingsWatch) skipAhead(ctx context.Context, box *watchedBox) error { + listed, err := sdk.Boxes().List(ctx) + if err != nil { + return convertSDKError(err) + } + if listed == nil { + return output.ErrAPI(0, "could not list boxes") + } + + for _, listedBox := range *listed { + if listedBox.Id == box.id { + cursor, err := watchCursor(listedBox.PostingChangesUrl, "") + if err != nil { + return err + } + box.cursor = cursor + } + } + + return nil +} + +func (w *postingsWatch) report(ctx context.Context, event watchEvent, box *watchedBox, posting *generated.Posting) { + if w.finished() || !w.changes[event.Change] { + return + } + + event.Box = watchEventBox{ID: box.id, Kind: box.kind, Name: box.name} + if posting != nil { + event.Posting = posting + event.ThreadID = resolvePostingTopicID(*posting) + } + w.reported++ + + switch { + case w.asyncScript != "": + w.spawnScript(ctx, event) + case w.syncScript != "": + w.lastScriptExit = w.runScript(ctx, event) + case w.styled: + fmt.Fprintln(w.out, watchLine(event)) + default: + w.writeJSON(event) + } +} + +func (w *postingsWatch) finished() bool { + return w.exitOnFirst && w.reported > 0 +} + +// spawnScript starts the script and leaves it to get on with it. Whether it worked, +// and whether it overlaps with the next one, is the script's business — and it outlives +// the watch, so interrupting `hey` doesn't cut a script off halfway. +// +// Only asyncScriptLimit of them run at once, though: a --since catch-up carries thousands +// of changes, and a slow script would have a process per change all fighting for the +// machine. Once they're all busy the watch waits for one to finish — or for an interrupt, +// which drops the change rather than hanging on a script that never ends. +func (w *postingsWatch) spawnScript(ctx context.Context, event watchEvent) { + command, err := w.scriptCommand(context.WithoutCancel(ctx), w.asyncScript, event) + if err != nil { + fmt.Fprintf(w.errOut, "warning: could not run %q: %v\n", w.asyncScript, err) + return + } + + select { + case w.running <- struct{}{}: + case <-ctx.Done(): + return + } + + if err := command.Start(); err != nil { + <-w.running + fmt.Fprintf(w.errOut, "warning: could not run %q: %v\n", w.asyncScript, err) + return + } + + go func() { + _ = command.Wait() + <-w.running + }() +} + +func (w *postingsWatch) runScript(ctx context.Context, event watchEvent) int { + command, err := w.scriptCommand(ctx, w.syncScript, event) + if err != nil { + fmt.Fprintf(w.errOut, "warning: could not run %q: %v\n", w.syncScript, err) + return 1 + } + + if err := command.Run(); err != nil { + var exit *exec.ExitError + if errors.As(err, &exit) { + fmt.Fprintf(w.errOut, "warning: %q exited %d\n", w.syncScript, exit.ExitCode()) + return exit.ExitCode() + } + + fmt.Fprintf(w.errOut, "warning: could not run %q: %v\n", w.syncScript, err) + return 1 + } + + return 0 +} + +// scriptCommand hands the event over twice: as JSON on the script's stdin, for jq, and +// as environment variables, for a one-liner that only wants to know what happened. +func (w *postingsWatch) scriptCommand(ctx context.Context, script string, event watchEvent) (*exec.Cmd, error) { + payload, err := json.Marshal(event) + if err != nil { + return nil, err + } + + command := shellCommand(ctx, script) + command.Stdin = bytes.NewReader(append(payload, '\n')) + command.Stdout = w.out + command.Stderr = w.errOut + command.Env = append(os.Environ(), event.environment()...) + + return command, nil +} + +func (w *postingsWatch) writeJSON(event watchEvent) { + payload, err := json.Marshal(event) + if err != nil { + fmt.Fprintf(w.errOut, "warning: could not write a change: %v\n", err) + return + } + + fmt.Fprintln(w.out, string(payload)) +} + +func shellCommand(ctx context.Context, script string) *exec.Cmd { + if runtime.GOOS == "windows" { + return exec.CommandContext(ctx, "cmd", "/c", script) //nolint:gosec // G204: running the command the caller asked for is the point + } + + return exec.CommandContext(ctx, "sh", "-c", script) //nolint:gosec // G204: running the command the caller asked for is the point +} + +func (e watchEvent) environment() []string { + environment := []string{ + "HEY_CHANGE=" + e.Change, + "HEY_AT=" + e.At, + "HEY_BOX_ID=" + strconv.FormatInt(e.Box.ID, 10), + "HEY_BOX_KIND=" + e.Box.Kind, + "HEY_BOX_NAME=" + e.Box.Name, + "HEY_POSTING_ID=" + strconv.FormatInt(e.PostingID, 10), + } + if e.ThreadID != 0 { + environment = append(environment, "HEY_THREAD_ID="+strconv.FormatInt(e.ThreadID, 10)) + } + + return environment +} + +func watchLine(event watchEvent) string { + description := fmt.Sprintf("posting %d", event.PostingID) + if event.Posting != nil { + description = fmt.Sprintf("%s — %s (thread %d)", event.Posting.Creator.Name, truncate(event.Posting.Summary, 50), event.ThreadID) + } + + return fmt.Sprintf("%s %-8s %-24s %s", event.At, event.Change, event.Box.Name, description) +} + +func watchTime(at time.Time) string { + if at.IsZero() { + return "" + } + + return at.UTC().Format("2006-01-02T15:04:05.000Z") +} diff --git a/internal/cmd/watch_test.go b/internal/cmd/watch_test.go new file mode 100644 index 00000000..9315ed53 --- /dev/null +++ b/internal/cmd/watch_test.go @@ -0,0 +1,428 @@ +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + + actioncable "github.com/basecamp/actioncable-go" + + "github.com/basecamp/hey-sdk/go/pkg/generated" + + "github.com/basecamp/hey-cli/internal/auth" +) + +func TestWatchedChanges(t *testing.T) { + command := newWatchCommand() + changes, err := command.watchedChanges() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !changes["added"] || !changes["updated"] || !changes["deleted"] { + t.Errorf("changes = %v, want every change by default", changes) + } + + command.events = []string{"Added", " deleted"} + changes, err = command.watchedChanges() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !changes["added"] || !changes["deleted"] || changes["updated"] { + t.Errorf("changes = %v, want added and deleted only", changes) + } + + command.events = []string{"moved"} + if _, err := command.watchedChanges(); err == nil { + t.Error("expected an error for an unknown event") + } + + command.events = nil + if _, err := command.watchedChanges(); err == nil { + t.Error("expected an error when no events are left to watch") + } +} + +func TestWatchRunFlagsAreEitherOr(t *testing.T) { + t.Setenv("HEY_TOKEN", "test-token") + t.Setenv("HEY_NO_KEYRING", "1") + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + + root := newRootCmd() + root.SetOut(&bytes.Buffer{}) + root.SetErr(&bytes.Buffer{}) + root.SetArgs([]string{"watch", "--run-async", "./notify.sh", "--run-sync", "./triage.sh"}) + + err := root.Execute() + if err == nil { + t.Fatal("expected an error when both run flags are given") + } + if !strings.Contains(err.Error(), "either --run-async or --run-sync") { + t.Errorf("error = %q, want it to name both flags as a choice", err.Error()) + } +} + +func TestWatchingBox(t *testing.T) { + imbox := generated.Box{Id: 24088, Kind: "imbox", Name: "Imbox"} + feed := generated.Box{Id: 24089, Kind: "feedbox", Name: "The Feed"} + + command := newWatchCommand() + if !command.watching(imbox) || !command.watching(feed) { + t.Error("every box should be watched when --box isn't given") + } + + command.boxes = []string{"IMBOX"} + if !command.watching(imbox) || command.watching(feed) { + t.Error("--box imbox should match the imbox by kind, case insensitively") + } + + command.boxes = []string{"The Feed"} + if !command.watching(feed) || command.watching(imbox) { + t.Error("--box should match a box by name") + } + + command.boxes = []string{"24089"} + if !command.watching(feed) || command.watching(imbox) { + t.Error("--box should match a box by ID") + } +} + +func TestWatchCursor(t *testing.T) { + changesURL := "https://app.hey.com/boxes/24088/postings/changes.json?since=2026-08-18T09%3A00%3A00.000Z&v=2" + + cursor, err := watchCursor(changesURL, "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cursor.Since != "2026-08-18T09:00:00.000Z" || cursor.Version != "2" { + t.Errorf("cursor = %+v, want the server's own since and version", cursor) + } + + cursor, err = watchCursor(changesURL, "2026-08-17T08:30:00Z") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cursor.Since != "2026-08-17T08:30:00.000Z" { + t.Errorf("since = %q, want --since to move it back", cursor.Since) + } + if cursor.Version != "2" { + t.Errorf("version = %q, want it to survive --since", cursor.Version) + } + + cursor, err = watchCursor(changesURL, "2026-08-17") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cursor.Since != "2026-08-17T00:00:00.000Z" { + t.Errorf("since = %q, want a bare date read as midnight", cursor.Since) + } + + if _, err := watchCursor(changesURL, "last tuesday"); err == nil { + t.Error("expected an error for a --since we can't read") + } +} + +func newTestWatch(changes ...string) (*postingsWatch, *bytes.Buffer) { + watched := map[string]bool{} + for _, change := range changes { + watched[change] = true + } + + out := &bytes.Buffer{} + return &postingsWatch{ + boxes: map[int64]*watchedBox{24088: {id: 24088, kind: "imbox", name: "Imbox"}}, + changes: watched, + out: out, + errOut: &bytes.Buffer{}, + catchUp: make(chan struct{}, 1), + unread: map[int64]bool{}, + running: make(chan struct{}, asyncScriptLimit), + }, out +} + +func TestWatchReportsJSONPerPosting(t *testing.T) { + watch, out := newTestWatch("added", "updated", "deleted") + posting := &generated.Posting{Id: 9001, AppUrl: "https://app.hey.com/topics/5511"} + + watch.report(context.Background(), watchEvent{Change: "added", At: "2026-08-18T09:14:22.031Z", PostingID: 9001}, watch.boxes[24088], posting) + watch.report(context.Background(), watchEvent{Change: "deleted", At: "2026-08-18T09:15:00.000Z", PostingID: 9003}, watch.boxes[24088], nil) + + lines := strings.Split(strings.TrimSpace(out.String()), "\n") + if len(lines) != 2 { + t.Fatalf("wrote %d lines, want one per change: %q", len(lines), out.String()) + } + + var added watchEvent + if err := json.Unmarshal([]byte(lines[0]), &added); err != nil { + t.Fatalf("first line isn't JSON: %v", err) + } + if added.Change != "added" || added.PostingID != 9001 || added.ThreadID != 5511 { + t.Errorf("added = %+v, want posting 9001 on thread 5511", added) + } + if added.Box.Kind != "imbox" || added.Box.ID != 24088 { + t.Errorf("box = %+v, want the imbox", added.Box) + } + if added.Posting == nil { + t.Error("an added posting should carry the posting itself") + } + + var deleted watchEvent + if err := json.Unmarshal([]byte(lines[1]), &deleted); err != nil { + t.Fatalf("second line isn't JSON: %v", err) + } + if deleted.Posting != nil { + t.Error("a deleted posting is gone, so there's nothing to carry") + } + if deleted.ThreadID != 0 { + t.Errorf("thread = %d, want none for a deleted posting", deleted.ThreadID) + } +} + +func TestWatchSkipsChangesItIsntWatching(t *testing.T) { + watch, out := newTestWatch("added") + + watch.report(context.Background(), watchEvent{Change: "updated", PostingID: 9002}, watch.boxes[24088], nil) + + if out.Len() != 0 { + t.Errorf("wrote %q, want nothing for a change outside --events", out.String()) + } +} + +func TestWatchExitOnFirstReportsOnce(t *testing.T) { + watch, out := newTestWatch("added") + watch.exitOnFirst = true + + watch.report(context.Background(), watchEvent{Change: "added", PostingID: 9001}, watch.boxes[24088], nil) + watch.report(context.Background(), watchEvent{Change: "added", PostingID: 9002}, watch.boxes[24088], nil) + + if lines := strings.Count(out.String(), "\n"); lines != 1 { + t.Errorf("wrote %d lines, want only the first change", lines) + } + if !watch.finished() { + t.Error("the watch should be finished after its first change") + } +} + +func TestWatchRunsScriptSynchronously(t *testing.T) { + watch, out := newTestWatch("added") + watch.syncScript = `printf "%s %s\n" "$HEY_CHANGE" "$HEY_POSTING_ID"; cat` + + watch.report(context.Background(), watchEvent{Change: "added", PostingID: 9001}, watch.boxes[24088], nil) + + if !strings.Contains(out.String(), "added 9001") { + t.Errorf("script output = %q, want the change in its environment", out.String()) + } + if !strings.Contains(out.String(), `"kind":"imbox"`) { + t.Errorf("script output = %q, want the event JSON on its stdin", out.String()) + } + if watch.lastScriptExit != 0 { + t.Errorf("exit = %d, want 0", watch.lastScriptExit) + } +} + +func TestWatchKeepsGoingWhenAScriptFails(t *testing.T) { + watch, _ := newTestWatch("added") + watch.syncScript = "exit 3" + + watch.report(context.Background(), watchEvent{Change: "added", PostingID: 9001}, watch.boxes[24088], nil) + + if watch.lastScriptExit != 3 { + t.Errorf("exit = %d, want the script's own 3", watch.lastScriptExit) + } + if warning := watch.errOut.(*bytes.Buffer).String(); !strings.Contains(warning, "exited 3") { + t.Errorf("stderr = %q, want the failure reported", warning) + } +} + +func TestWatchEventEnvironment(t *testing.T) { + event := watchEvent{ + Change: "added", + At: "2026-08-18T09:14:22.031Z", + Box: watchEventBox{ID: 24088, Kind: "imbox", Name: "Imbox"}, + PostingID: 9001, + ThreadID: 5511, + } + + environment := strings.Join(event.environment(), "\n") + for _, want := range []string{"HEY_CHANGE=added", "HEY_BOX_ID=24088", "HEY_BOX_KIND=imbox", "HEY_BOX_NAME=Imbox", "HEY_POSTING_ID=9001", "HEY_THREAD_ID=5511", "HEY_AT=2026-08-18T09:14:22.031Z"} { + if !strings.Contains(environment, want) { + t.Errorf("environment = %q, want %s", environment, want) + } + } +} + +func TestWatchReadsChangesWhenNotified(t *testing.T) { + t.Setenv("HEY_TOKEN", "test-token") + + var requested []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requested = append(requested, r.URL.String()) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Link", `<`+r.URL.Path+`?since=2026-08-18T09%3A14%3A22.031Z&v=2>; rel="next"`) + _, _ = w.Write([]byte(`{"added":[{"id":9001,"kind":"topic","box_id":24088,"app_url":"https://app.hey.com/topics/5511"}]}`)) + })) + defer server.Close() + initSDK(auth.NewManager(server.URL, server.Client(), t.TempDir()), server.URL) + + watch, out := newTestWatch("added", "updated", "deleted") + cursor, err := watchCursor(server.URL+"/boxes/24088/postings/changes.json?since=2026-08-18T09%3A00%3A00.000Z&v=2", "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + watch.boxes[24088].cursor = cursor + + if err := watch.read(context.Background(), actioncable.Message(`{"change":"upsert","box_id":24088}`)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(requested) != 1 { + t.Fatalf("requests = %v, want one read of the changes feed", requested) + } + if !strings.Contains(requested[0], "since=2026-08-18T09%3A00%3A00.000Z") { + t.Errorf("requested %q, want the box's own cursor", requested[0]) + } + + var event watchEvent + if err := json.Unmarshal([]byte(strings.TrimSpace(out.String())), &event); err != nil { + t.Fatalf("output isn't JSON: %q", out.String()) + } + if event.Change != "added" || event.PostingID != 9001 || event.ThreadID != 5511 { + t.Errorf("event = %+v, want added posting 9001 on thread 5511", event) + } + if watch.boxes[24088].cursor.Since != "2026-08-18T09:14:22.031Z" { + t.Errorf("cursor = %+v, want it moved to where the feed left off", watch.boxes[24088].cursor) + } + + if err := watch.read(context.Background(), actioncable.Message(`{"change":"upsert","box_id":99999}`)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(requested) != 1 { + t.Errorf("requests = %v, want none for a box we aren't watching", requested) + } +} + +func TestWatchReadsAgainAfterAFailedRead(t *testing.T) { + t.Setenv("HEY_TOKEN", "test-token") + + broken := true + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if broken { + http.Error(w, "boom", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Link", `<`+r.URL.Path+`?since=2026-08-18T09%3A14%3A22.031Z&v=2>; rel="next"`) + _, _ = w.Write([]byte(`{"added":[{"id":9001,"kind":"topic","box_id":24088,"app_url":"https://app.hey.com/topics/5511"}]}`)) + })) + defer server.Close() + initSDK(auth.NewManager(server.URL, server.Client(), t.TempDir()), server.URL) + + watch, out := newTestWatch("added") + cursor, err := watchCursor(server.URL+"/boxes/24088/postings/changes.json?since=2026-08-18T09%3A00%3A00.000Z&v=2", "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + watch.boxes[24088].cursor = cursor + + if err := watch.read(context.Background(), actioncable.Message(`{"change":"upsert","box_id":24088}`)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if !watch.unread[24088] { + t.Error("a box whose read failed should be waiting to be read again") + } + if watch.retry == nil { + t.Error("a failed read should arm a retry") + } + if watch.backoff != firstWatchRetry { + t.Errorf("backoff = %s, want the first retry to wait %s", watch.backoff, firstWatchRetry) + } + if watch.boxes[24088].cursor.Since != "2026-08-18T09:00:00.000Z" { + t.Errorf("cursor = %+v, want it left where it was so the retry picks the change up", watch.boxes[24088].cursor) + } + + broken = false + if err := watch.readUnreadBoxes(context.Background()); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(watch.unread) != 0 { + t.Errorf("unread = %v, want the box cleared once it was read", watch.unread) + } + if watch.backoff != 0 { + t.Errorf("backoff = %s, want it reset once every box was read", watch.backoff) + } + if !strings.Contains(out.String(), `"posting_id":9001`) { + t.Errorf("output = %q, want the change the failed read missed", out.String()) + } +} + +func TestWatchClosedSubscriptionIsOnlyFineWhenItWasInterrupted(t *testing.T) { + watch, _ := newTestWatch("added") + + interrupted, interrupt := context.WithCancel(context.Background()) + interrupt() + if err := watch.closedError(interrupted); err != nil { + t.Errorf("error = %v, want an interrupted watch to end cleanly", err) + } + + err := watch.closedError(context.Background()) + if err == nil { + t.Fatal("a connection that went away for good should be reported") + } + if !strings.Contains(err.Error(), "hung up") { + t.Errorf("error = %q, want it to say the server hung up", err.Error()) + } + + watch.rejected.Store(true) + err = watch.closedError(context.Background()) + if err == nil || !strings.Contains(err.Error(), "turned this subscription down") { + t.Errorf("error = %v, want a rejected subscription reported as an auth failure", err) + } +} + +func TestWatchRunsBoundedAsyncScripts(t *testing.T) { + watch, _ := newTestWatch("added") + ran := t.TempDir() + "/ran" + watch.asyncScript = "echo $HEY_POSTING_ID >> " + ran + // Overlapping scripts share whatever the watch writes to, and a test's buffer isn't + // the file descriptor they'd be handed in a terminal. + watch.out, watch.errOut = io.Discard, io.Discard + + for posting := range int64(asyncScriptLimit * 2) { + watch.report(context.Background(), watchEvent{Change: "added", PostingID: 9000 + posting}, watch.boxes[24088], nil) + if len(watch.running) > asyncScriptLimit { + t.Fatalf("%d scripts running at once, want no more than %d", len(watch.running), asyncScriptLimit) + } + } + + for range asyncScriptLimit { + watch.running <- struct{}{} + } + + lines, err := os.ReadFile(ran) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if count := strings.Count(string(lines), "\n"); count != asyncScriptLimit*2 { + t.Errorf("%d scripts ran, want one per change", count) + } +} + +func TestAskForCatchUpNeverBlocks(t *testing.T) { + watch, _ := newTestWatch("added") + + watch.askForCatchUp() + watch.askForCatchUp() + + select { + case <-watch.catchUp: + default: + t.Fatal("a catch-up should be waiting") + } +} diff --git a/nix/package.nix b/nix/package.nix index edfbe3fb..de616ce8 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -18,7 +18,7 @@ buildGoModule.override { inherit go; } (finalAttrs: { # To update: run `make update-nix-hash` (Docker). It rewrites this quoted # value in place, so keep it a string literal rather than lib.fakeHash. - vendorHash = "sha256-81G63fNdrq1TbLx5LCL9TrY69G1ejMupEBASazW5sjk="; + vendorHash = "sha256-U+2R7AemoMHdQjf+XjrX675I4+GRPBXkhlFMv1OITmk="; subPackages = [ "cmd/hey" ]; diff --git a/skills/hey/SKILL.md b/skills/hey/SKILL.md index c24007e5..cbf15506 100644 --- a/skills/hey/SKILL.md +++ b/skills/hey/SKILL.md @@ -158,6 +158,8 @@ hey boxes --quiet --jq '.[].name' | Complete todo | `hey todo complete 123` | | Uncomplete todo | `hey todo uncomplete 123` | | Delete todo | `hey todo delete 123` | +| Wait for new mail | `hey watch --box imbox --exit-on-first` | +| Follow every change | `hey watch` | | Mark as seen | `hey seen 12345` | | Mark as unseen | `hey unseen 12345` | | Move email threads | `hey move 12345 --to feed` | @@ -374,6 +376,30 @@ hey stop-ignoring 12345 67890 # Stop ignoring multiple threads Takes box item IDs (the `id` field from `hey box --json`). Ignored threads remain in their box; new replies do not bring them back to your attention. `hey stop-ignoring` reverses the action. +### Email - Watching for changes + +```bash +hey watch # Follow every box until interrupted +hey watch --box imbox # Follow one box (repeatable, by name or ID) +hey watch --events added,deleted # Only these changes (added, updated, deleted) +hey watch --exit-on-first # Wait for one change, print it, exit +hey watch --timeout 30m # Give up waiting after a while +hey watch --since 2026-03-15 # Report changes since then first, then follow +hey watch --run-sync ./triage.sh # Run a command per change instead of printing +``` + +Long-running, and driven by a websocket rather than polling — never poll `hey box` in a +loop when this will do. Writes one JSON object per changed posting to stdout, one per +line, instead of the usual envelope: `{"change": "added", "at": ..., "box": {"id", "kind", +"name"}, "posting_id": ..., "thread_id": ..., "posting": {...}}`. Use `thread_id` with +`hey threads`. A deleted posting carries no `posting` or `thread_id`. + +To drive a command per change, choose one of two behaviours — passing both is an error. +`--run-async` spawns the command and moves on, so a slow one never holds up the watch and +two can overlap; `--run-sync` waits for each and runs them in order. Both get the JSON on +stdin and the fields as `HEY_CHANGE`, `HEY_AT`, `HEY_BOX_ID`, `HEY_BOX_KIND`, +`HEY_BOX_NAME`, `HEY_POSTING_ID` and `HEY_THREAD_ID`, and both take over stdout. + ### Drafts ```bash