feat: migrate mail watch to event consume#1650
Open
chumbleyneria wants to merge 7 commits into
Open
Conversation
|
|
github-actions
Bot
added
domain/mail
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds subscription-scoped identity across the event protocol, bus, and consume flow, extracts the ChangesSubscription-scoped event consume and mail event handler
Sequence Diagram(s)sequenceDiagram
participant CLI as cmd/event consume or mail +watch
participant Runner as consumecli.Run
participant Consume as consume.Run
participant Bus as event bus
CLI->>Runner: Run(opts)
Runner->>Runner: NormalizeParams, ComputeSubscriptionID
Runner->>Runner: Preflight checks
Runner->>Consume: Run(..., subscriptionID)
Consume->>Bus: Hello{EventKey, SubscriptionID}
Bus-->>Consume: HelloAck
loop events
Bus-->>Consume: RawEvent
Consume->>Consume: Match → Process → Sink
end
Consume->>Bus: PreShutdownCheck{EventKey, SubscriptionID}
Bus-->>Consume: PreShutdownAck
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
🧹 Nitpick comments (1)
internal/event/protocol/messages_test.go (1)
20-29: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winPin the new
SubscriptionIDconstructor behavior.Line 20 and Line 29 pass
"", so this constructor test won’t catch a future regression where the new parameter is ignored.Suggested test update
- if got := NewHello(1, "k", []string{"t"}, "v1", ""); got.Type != MsgTypeHello { + if got := NewHello(1, "k", []string{"t"}, "v1", "k:sub"); got.Type != MsgTypeHello || got.SubscriptionID != "k:sub" { t.Errorf("NewHello.Type = %q, want %q", got.Type, MsgTypeHello) } @@ - if got := NewPreShutdownCheck("k", ""); got.Type != MsgTypePreShutdownCheck || got.EventKey != "k" { + if got := NewPreShutdownCheck("k", "k:sub"); got.Type != MsgTypePreShutdownCheck || got.EventKey != "k" || got.SubscriptionID != "k:sub" { t.Errorf("NewPreShutdownCheck mismatch: %+v", got) }As per coding guidelines, “Every behavior change needs a test alongside the change.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/event/protocol/messages_test.go` around lines 20 - 29, The constructor tests for NewHello and NewPreShutdownCheck still pass an empty SubscriptionID, so they won’t detect regressions where the new parameter is ignored. Update the assertions in messages_test.go to use a non-empty SubscriptionID and verify it is preserved on the returned message structs, alongside the existing checks for Type, FirstForKey, Seq, and EventKey.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@events/mail/message_received_test.go`:
- Around line 125-146: The test name claims to cover the event-format branch,
but `TestProcessOutputDirForcesFullFetchEvenForEventFormat` currently uses
`processMessageReceived` with `msg_format` set to `minimal`, so it does not
exercise the `event` path. Update the test data to use `msg_format: "event"` (or
split this into a separate case) so the assertions verify the special
`watch_output_dir_full` behavior in the actual event-format flow and protect the
`processMessageReceived` branch from regressions.
In `@events/mail/message_received.go`:
- Around line 163-171: Wrap every bare JSON decode failure in
message_received.go with a typed errs problem instead of returning the raw
json.Unmarshal error; update the decode paths in the profile, mailbox/message,
and related response handling code so they use
errs.NewInternalError(errs.SubtypeInvalidResponse, ...).WithCause(err) while
leaving existing typed CallAPI errors unchanged. Use the existing decode points
in message_received.go (for example the response parsing around the profile
fetch and the other listed unmarshal sites) to locate and replace the direct
returns, preserving the underlying cause with .WithCause(err).
In `@events/mail/register.go`:
- Around line 55-70: The mail event registration currently uses a single static
mailScopes list in register.go, which causes consumecli preflight to demand
fetch-only permissions even when processMessageReceived runs in no-fetch mode
for msg_format=event. Update the scope definition around the KeyDefinition in
register.go so preflight can distinguish baseline subscribe scopes from
fetch-only scopes, or make the scope check aware of the requested params by
using normalizeParams/processMessageReceived mode to avoid rejecting
least-privilege runs like mailbox=alice@example.com,msg_format=event.
In `@events/whiteboard/preconsume.go`:
- Around line 42-48: The new cleanup contract in the preconsume unsubscribe path
needs an error-path test. Update the existing test coverage around cleanup() in
preconsume_test to add a case where rt.CallAPI fails for the unsubscribePath and
verify the returned error is preserved unchanged. Assert the typed problem
metadata with errs.ProblemOf, including category, subtype, and param, and also
check the cause is preserved rather than relying on message matching.
In `@internal/event/bus/handle_hello_test.go`:
- Around line 100-105: The HelloAck read in the handleHello tests can block
indefinitely, so add a timeout around the blocking ReadString('\n') call before
waiting on the later select. Update the test logic in the handleHello test
helpers/blocks to set a read deadline on client or perform the read in a
goroutine with timeout protection, so a missing ack fails quickly instead of
hanging. Use the same approach in both test cases that currently read the
HelloAck directly.
In `@internal/event/consume/consume.go`:
- Around line 106-109: The handshake error path in doHello handling currently
wraps everything with fmt.Errorf, which downgrades typed command errors. Update
the error handling in consume/run flow so any already-typed handshake error is
returned unchanged, and only untyped failures are converted using the
appropriate errs.* constructor with .WithCause(err). Keep the change localized
around doHello and the Run boundary so error classification and unwrapping
remain intact.
In `@internal/event/consume/loop.go`:
- Around line 202-205: Populate the full event.RawEvent before hook invocation
in the consume loop so Match and Process can see EventID and SourceTime, not
just EventType and Payload. Update the RawEvent construction in the event
handling path in loop.go to copy the metadata from protocol.Event into
event.RawEvent, and make sure the same change is applied wherever the same
initializer is duplicated. Add a small regression test around the consume
loop/hook path that verifies EventID and SourceTime are preserved and available
to a hook implementation.
In `@internal/event/consume/sink.go`:
- Around line 30-33: The MkdirAll failure path in the output-directory setup is
returning an untyped fmt.Errorf, so callers lose the expected file-I/O error
metadata. Update the opts.OutputDir handling in sink.go to validate the path
first if needed, then wrap vfs.MkdirAll errors with
errs.NewInternalError(errs.SubtypeFileIO, ...) and attach the underlying cause
with .WithCause(err) so the existing error category and unwrap/is behavior are
preserved.
In `@internal/event/consumecli/runner.go`:
- Around line 306-312: The output-dir validation in validate.SafeOutputPath
handling is losing the real path-validation failure by attaching
ErrOutputDirUnsafe as the cause instead of the returned err. Update the error
construction in runner.go so the errs.NewValidationError chain keeps the
sentinel in the message if needed, but calls WithCause(err) to preserve the
concrete rejection reason for errors.Is/errors.Unwrap. Use the existing
validate.SafeOutputPath and WithCause chain as the location to fix.
In `@shortcuts/mail/mail_watch.go`:
- Around line 182-202: The compatibility envelope identity is derived from
runtime.As() while runMailWatchConsume is forced to use core.AsUser, so the two
can diverge. Update the mail_watch flow in the runMailWatchConsume call site to
build consume.OutputEnvelope with the same forced identity used by
IdentityOverride, ensuring Envelope.Identity and the delegated consume identity
stay in sync.
- Around line 24-25: The new `vfs.MkdirAll` usage in `mail_watch.go` violates
the `shortcuts-no-vfs` depguard rule because `shortcuts/` code must not import
`internal/vfs` directly. Remove the `vfs` dependency from `mailWatch`-related
code and route directory creation through the shortcut/runtime file abstraction
used elsewhere in `shortcuts/`, updating any related calls in the referenced
block so the logic stays the same without importing `vfs`.
---
Nitpick comments:
In `@internal/event/protocol/messages_test.go`:
- Around line 20-29: The constructor tests for NewHello and NewPreShutdownCheck
still pass an empty SubscriptionID, so they won’t detect regressions where the
new parameter is ignored. Update the assertions in messages_test.go to use a
non-empty SubscriptionID and verify it is preserved on the returned message
structs, alongside the existing checks for Type, FirstForKey, Seq, and EventKey.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: dddc3ada-1274-4820-90c0-7feab6838087
📒 Files selected for processing (46)
cmd/event/consume.gocmd/event/format_helpers_test.gocmd/event/schema.gocmd/event/schema_test.gocmd/event/status.goevents/mail/message_received.goevents/mail/message_received_test.goevents/mail/register.goevents/minutes/preconsume.goevents/register.goevents/vc/preconsume.goevents/whiteboard/preconsume.gointernal/event/bus/bus.gointernal/event/bus/bus_shutdown_test.gointernal/event/bus/conn.gointernal/event/bus/conn_test.gointernal/event/bus/handle_hello_test.gointernal/event/bus/hub.gointernal/event/bus/hub_observability_test.gointernal/event/bus/hub_publish_race_test.gointernal/event/bus/hub_test.gointernal/event/consume/consume.gointernal/event/consume/consume_test.gointernal/event/consume/fingerprint.gointernal/event/consume/fingerprint_test.gointernal/event/consume/handshake.gointernal/event/consume/handshake_test.gointernal/event/consume/loop.gointernal/event/consume/loop_test.gointernal/event/consume/shutdown.gointernal/event/consume/shutdown_test.gointernal/event/consume/sink.gointernal/event/consume/sink_test.gointernal/event/consumecli/appmeta_err.gointernal/event/consumecli/console_url.gointernal/event/consumecli/runner.gointernal/event/consumecli/sigpipe_unix.gointernal/event/consumecli/sigpipe_windows.gointernal/event/consumecli/suggestions.gointernal/event/protocol/codec_test.gointernal/event/protocol/messages.gointernal/event/protocol/messages_test.gointernal/event/types.goshortcuts/mail/mail_watch.goshortcuts/mail/mail_watch_test.goskills/lark-mail/references/lark-mail-watch.md
Comment on lines
+42
to
+48
| return func() error { | ||
| cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) | ||
| defer cancel() | ||
| _, _ = rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body) | ||
| if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil { | ||
| return err | ||
| } | ||
| return nil |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add an unsubscribe failure test for the new cleanup contract.
This change makes cleanup() observable, but the provided downstream test at events/whiteboard/preconsume_test.go:99-144 still only verifies that unsubscribe is called and ignores the returned error. Please add a case where rt.CallAPI fails on the unsubscribe path and assert that cleanup() preserves the typed problem metadata/cause unchanged. As per coding guidelines, "Every behavior change needs a test alongside the change" and "**/*_test.go: Error-path tests must assert typed metadata via errs.ProblemOf (category / subtype / param) and cause preservation, not message substrings alone."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@events/whiteboard/preconsume.go` around lines 42 - 48, The new cleanup
contract in the preconsume unsubscribe path needs an error-path test. Update the
existing test coverage around cleanup() in preconsume_test to add a case where
rt.CallAPI fails for the unsubscribePath and verify the returned error is
preserved unchanged. Assert the typed problem metadata with errs.ProblemOf,
including category, subtype, and param, and also check the cause is preserved
rather than relying on message matching.
Source: Coding guidelines
Comment on lines
+100
to
+105
| // Read the HelloAck from client side to let handleHello complete. | ||
| clientReader := bufio.NewReader(client) | ||
| ackLine, err := clientReader.ReadString('\n') | ||
| if err != nil { | ||
| t.Fatalf("failed to read HelloAck: %v", err) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Put a deadline around the HelloAck read.
Both tests call ReadString('\n') before the timeout select, so a regression that stops handleHello from writing the ack will hang the test until the package timeout instead of failing in 3 seconds. Set a read deadline on client (or read in a goroutine with a timeout) before the blocking read.
Suggested fix
+ client.SetReadDeadline(time.Now().Add(3 * time.Second))
ackLine, err := clientReader.ReadString('\n')
if err != nil {
t.Fatalf("failed to read HelloAck: %v", err)
}
+ _ = client.SetReadDeadline(time.Time{})Also applies to: 166-171
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/event/bus/handle_hello_test.go` around lines 100 - 105, The HelloAck
read in the handleHello tests can block indefinitely, so add a timeout around
the blocking ReadString('\n') call before waiting on the later select. Update
the test logic in the handleHello test helpers/blocks to set a read deadline on
client or perform the read in a goroutine with timeout protection, so a missing
ack fails quickly instead of hanging. Use the same approach in both test cases
that currently read the HelloAck directly.
Comment on lines
+202
to
+205
| raw := &event.RawEvent{ | ||
| EventType: evt.EventType, | ||
| Payload: evt.Payload, | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Populate the full RawEvent before invoking hooks.
protocol.Event already has EventID and SourceTime, and event.RawEvent exposes those fields, but this initializer drops them. Any Match or Process implementation that relies on event metadata will see zero values even though the wire message has the data.
🩹 Proposed fix
raw := &event.RawEvent{
+ EventID: evt.EventID,
EventType: evt.EventType,
+ SourceTime: evt.SourceTime,
Payload: evt.Payload,
}Please add a small regression test for metadata propagation alongside the fix. As per coding guidelines, “Every behavior change needs a test alongside the change.”
Also applies to: 207-210
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/event/consume/loop.go` around lines 202 - 205, Populate the full
event.RawEvent before hook invocation in the consume loop so Match and Process
can see EventID and SourceTime, not just EventType and Payload. Update the
RawEvent construction in the event handling path in loop.go to copy the metadata
from protocol.Event into event.RawEvent, and make sure the same change is
applied wherever the same initializer is duplicated. Add a small regression test
around the consume loop/hook path that verifies EventID and SourceTime are
preserved and available to a hook implementation.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/event/consume/consume_test.go`:
- Around line 125-133: Update the error-path assertions in the consume test to
check typed metadata instead of relying on multiple substrings from err.Error().
Use errs.ProblemOf(err) to assert the expected category and subtype, and use
errors.As with *errs.ValidationError to verify the failed param value. Keep only
one message-based assertion for the CLI guidance text, and reference the
existing consume test/error handling path so the checks stay tied to the right
validation failure.
In `@internal/event/consume/consume.go`:
- Around line 238-239: The invalid enum value path in consume validation is
returning a plain formatted error instead of a typed validation error. Update
the error handling around the EventKey parameter check in the consume flow to
use errs.NewValidationError(errs.SubtypeInvalidArgument, ...) with the same
message, chain .WithParam(p.Name), and move the schema guidance text into
.WithHint(...) so it matches the other command-facing validation paths.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c2616cdc-bb45-4018-b010-f5edc107b100
📒 Files selected for processing (2)
internal/event/consume/consume.gointernal/event/consume/consume_test.go
Change-Type: ci-fix
Change-Type: ci-fix
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Tests
Summary by CodeRabbit
mail.user_mailbox.event.message_received_v1support and unifiedmail +watchvia the sharedevent consumeflow, including--max-events/--timeoutand optional legacy JSON envelope output.event consumeworkflow and clarify output formatting.