feat(events): per-resource subscription identity + Match hook + mail - #1184
Closed
liuxinyanglxy wants to merge 21 commits into
Closed
feat(events): per-resource subscription identity + Match hook + mail#1184liuxinyanglxy wants to merge 21 commits into
liuxinyanglxy wants to merge 21 commits into
Conversation
… cleanup to func() error Change-Id: I23ba07c089b2c0eb3e42451d8118849a1c5f7a9d
…dedup Change-Id: Iaeb09c15c754c9e4ca8ce9784623f82f55fe308b
…nsumerInfo Add SubscriptionID field (omitempty) to Hello, PreShutdownCheck, and ConsumerInfo. Update NewHello and NewPreShutdownCheck constructors to accept subscriptionID param; patch all call sites with "" placeholder pending Tasks 7/8. Change-Id: I469a1fb0de2c31326513feef5693d8097dfc8415
Add subID field to Conn, update NewConn signature to accept it, add SubscriptionID() getter (falls back to EventKey when empty), update handleControlMessage PreShutdownCheck branch to use message SubscriptionID with EventKey fallback, and wire hello.SubscriptionID through handleHello. Change-Id: Iea9195dd2d343fd51c84317e4d7f2d5d8067d51d
Change-Id: I3cc81e8440656470f92f1b50074b840b0875c781
Change-Id: Ieb7c2237621a84721b6fc73967838d9caec2a837
…o Run Change-Id: Ifd43aa2a82c315ad3d496a39ce9f0dbdd43f3bb8
…sumeLoop Change-Id: Ib40b148c45f546ca112d30976749f42fae1f877a
…note Change-Id: I121ec0d3e7d50f3ebb728f47f1f54d89322e0143
Run keyDef.Match(raw, params) at the top of processAndOutput before any Process/sink work; drop the event (return false, nil) when it returns false. Move the RawEvent allocation above the Match guard so both Match and Process share the same struct. Three new unit tests cover drop, nil-accepts-all, and run-before-process ordering. Change-Id: I21571b3da91a9a918903b6fc3a6ff897eb83a747
Change-Id: I2f88774c58719af0419c471f40de1bf12242a5ab
Add a SUB column to the human-readable consumer table in `event status`. Renders the fingerprint hash suffix (after the colon) when present; falls back to "-" for legacy consumers where SubscriptionID is empty or equal to EventKey. Change-Id: Ia6271475e562844fc275f3751c64ca408ebda82c
Change-Id: Ie880d12e5dda73925ccaadc60ff8d3acda401dd6
Change-Id: I02cca113d50067e9a5964b896b2bc84210d2a35b
Change-Id: I3dcd481fa93ce8f5dbde1e065aa00c18330265c0
Change-Id: I99c64618900dde0b5420ce8a4f5c088960098e66
…ork hooks Adds events/mail/register.go with Keys() exposing the mail.user_mailbox.event.message_received_v1 EventKey: mailbox (SUB-KEY), folders, labels, msg-format params; wires NormalizeParams, PreConsume (subscribe/unsubscribe), Match, and Process hooks. Updates events/register.go to include mail.Keys() alongside im.Keys(). Note: Schema.Custom is used (not Native) because processMailEvent produces the complete output shape — Schema.Native is incompatible with Process per registry validation. Change-Id: Ibf48dc19dee5db65730810b0f2e4b5ebed73c4f0
Change-Id: I14636cae6a459cdf5a82455bc18022d5edd6c5ce
…ddress Change-Id: I03d0780aaac0fd7d59c1395df6912f5bb043f996
Change-Id: If46580dff10e15a56858157dbcdfb79f2c43d227
The mail message_received event's subscriber field is a
{user_ids:[{user_id,open_id,union_id}]} object per the Feishu SDK
(P2UserMailboxEventMessageReceivedV1Data.Subscriber), not a string.
Decoding it as string made Process fail at runtime with
'cannot unmarshal object into Go struct field .event.subscriber of
type string', silently dropping every mail event.
Change-Id: I91dcd7e82659e3962a2530fae8d5ee34fe40dbf6
📝 WalkthroughWalkthroughThis PR introduces comprehensive mail event support to the Lark CLI event system. It refactors the event bus to track subscriptions by stable subscription ID instead of event key, implements mail event processing with filtering and enrichment, and adds CLI display and documentation for the new feature. ChangesMail Event Support with Subscription-ID Tracking
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
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 |
github-actions
Bot
added
domain/mail
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (1)
cmd/event/schema_test.go (1)
147-169: ⚡ Quick winAssert
subscription_keystructurally, not by substring.The current check can false-pass if any other boolean field renders as
true. Unmarshal the payload and verify the specific param entry instead.Suggested fix
func TestSchema_JSON_IncludesSubscriptionKey(t *testing.T) { const syntheticKey = "test.evt_json" t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) }) @@ if err := runSchema(f, syntheticKey, true); err != nil { t.Fatalf("runSchema json: %v", err) } - if !strings.Contains(stdout.String(), `"subscription_key"`) { - t.Errorf("JSON output missing subscription_key field: %s", stdout.String()) - } - if !strings.Contains(stdout.String(), `true`) { - t.Errorf("JSON output missing subscription_key: true value: %s", stdout.String()) - } + var payload struct { + Params []struct { + Name string `json:"name"` + SubscriptionKey bool `json:"subscription_key"` + } `json:"params"` + } + if err := json.Unmarshal(stdout.Bytes(), &payload); err != nil { + t.Fatalf("output is not valid JSON: %v\n%s", err, stdout.String()) + } + if len(payload.Params) != 1 || payload.Params[0].Name != "mailbox" || !payload.Params[0].SubscriptionKey { + t.Errorf("unexpected params payload: %+v", payload.Params) + } }🤖 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 `@cmd/event/schema_test.go` around lines 147 - 169, The test TestSchema_JSON_IncludesSubscriptionKey currently asserts subscription_key by substring which can false-pass; instead parse the JSON output from stdout (the result of runSchema called with syntheticKey) into a suitable structure (e.g., map[string]interface{} or a struct matching the schema output), locate the params array entry for the param named "mailbox" (registered via eventlib.RegisterKey in the test) and assert that that param's "subscription_key" field exists and is true; update the assertions to use the unmarshalled value rather than string containment checks.
🤖 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/match.go`:
- Around line 31-33: The handler currently does json.Unmarshal(raw.Payload,
&env) and returns true on error (fail-open), which can route events to the wrong
mailbox when params["mailbox"] is set; change the logic in the mailbox filter to
fail-closed by returning false when json.Unmarshal returns an error (optionally
logging the unmarshal error and the offending raw.Payload and/or rejecting the
event), i.e., in the block around json.Unmarshal(raw.Payload, &env) switch the
return value on err from true to false so the event is rejected rather than
passed through to per-mailbox subscribers.
In `@internal/event/bus/conn.go`:
- Around line 150-156: In PreShutdownCheck (the Conn method where scope is
computed from m.SubscriptionID/m.EventKey), stop trusting the incoming message
identity and use the connection’s registered scope by calling c.SubscriptionID()
when computing scope for c.checkLastForKey; if m.SubscriptionID or m.EventKey
differ from c.SubscriptionID(), optionally emit a debug/warn log about the
mismatch before proceeding so checkLastForKey always evaluates the conn’s true
subscription scope.
In `@internal/event/consume/consume_test.go`:
- Around line 35-55: The test currently reconstructs the wrapper string with
fmt.Errorf instead of exercising the real wrapping logic; update
TestNormalizeParams_ErrorIsWrappedWithEventKey to either (A) invoke the actual
code path that performs the wrap (call the production wrapper used around
keyDef.NormalizeParams — the same function Run uses) so the returned err is the
wrapped error coming from that logic, or (B) extract the wrapping/formatting
into a helper (e.g., a new helper used by Run/NormalizeParams wrapper) and
assert that helper directly; then assert on the returned error from that real
wrapper (use strings.Contains or strings.HasPrefix on err.Error() for the prefix
"normalize params for "+keyDef.Key and use errors.Is/unwrap to verify the
underlying "simulated normalize failure") instead of building a new fmt.Errorf
in the test.
In `@internal/event/consume/consume.go`:
- Around line 61-67: Move the timeout-wrapping of the request context so it is
applied before calling keyDef.NormalizeParams to ensure NormalizeParams runs
under opts.Timeout; specifically, in the Consume flow wrap the incoming ctx with
the timeout (the existing opts.Timeout installation code) before invoking
keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params) and ensure the deferred
cancel is still called after NormalizeParams returns; adjust any references to
ctx passed to NormalizeParams, doHello, PreConsume, Match, and Process so they
all use the timeout-wrapped context.
In `@skills/lark-event/references/lark-event-mail.md`:
- Around line 117-122: Update the paragraph describing server-side subscription
keying to include the mailbox/resource dimension so it matches earlier
documentation: change the described key from "(app, user, event_type)" to
include "mailbox" (e.g., "(app, user, mailbox, event_type)"), and adjust the
explanations about overwrite/leak behavior and the warning about manual
unsubscribe so they refer to per-mailbox subscriptions and show that failed
unsubscribe leaks are per-mailbox (not global per-user) and that manual
unsubscribe can terminate a different consumer on the same mailbox; ensure the
text references the same "mailbox" terminology used earlier in this document.
In `@skills/lark-event/SKILL.md`:
- Around line 157-162: The current text incorrectly implies a single server
record per `(app, user, event_type)` and uses "reference-counted"; update the
wording to state that subscriptions are tracked by subscription scope (the
SubscriptionID / per-mailbox qualifier) so different mailbox/resource
subscriptions have distinct server records, and replace "reference-counted" with
a gentler description (e.g., server-side subscriptions are keyed by
SubscriptionID and are idempotent for the same scope), then rephrase the WARN
explanation to say the warning is informational and the next subscribe for the
same SubscriptionID self-heals rather than implying global shared state.
---
Nitpick comments:
In `@cmd/event/schema_test.go`:
- Around line 147-169: The test TestSchema_JSON_IncludesSubscriptionKey
currently asserts subscription_key by substring which can false-pass; instead
parse the JSON output from stdout (the result of runSchema called with
syntheticKey) into a suitable structure (e.g., map[string]interface{} or a
struct matching the schema output), locate the params array entry for the param
named "mailbox" (registered via eventlib.RegisterKey in the test) and assert
that that param's "subscription_key" field exists and is true; update the
assertions to use the unmarshalled value rather than string containment checks.
🪄 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: 7472c298-466d-44f3-95c3-5a11f7accdec
📒 Files selected for processing (42)
cmd/event/format_helpers_test.gocmd/event/schema.gocmd/event/schema_test.gocmd/event/status.goevents/mail/match.goevents/mail/match_test.goevents/mail/normalize.goevents/mail/normalize_test.goevents/mail/payload.goevents/mail/process.goevents/mail/process_test.goevents/mail/register.goevents/mail/register_test.goevents/register.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/protocol/codec_test.gointernal/event/protocol/messages.gointernal/event/protocol/messages_test.gointernal/event/types.goskills/lark-event/SKILL.mdskills/lark-event/references/lark-event-mail.mdskills/lark-mail/SKILL.mdskills/lark-mail/references/lark-mail-triage.mdskills/lark-mail/references/lark-mail-watch.md
💤 Files with no reviewable changes (1)
- skills/lark-mail/references/lark-mail-watch.md
Comment on lines
+31
to
+33
| if err := json.Unmarshal(raw.Payload, &env); err != nil { | ||
| return true // fail-open | ||
| } |
There was a problem hiding this comment.
Fail closed when a mailbox-filtered payload cannot be parsed.
Once params["mailbox"] is set, returning true on unmarshal failure can deliver an event to the wrong per-mailbox subscriber. Here the safer default is to reject the event rather than bypass the filter.
Suggested fix
if err := json.Unmarshal(raw.Payload, &env); err != nil {
- return true // fail-open
+ return false
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if err := json.Unmarshal(raw.Payload, &env); err != nil { | |
| return true // fail-open | |
| } | |
| if err := json.Unmarshal(raw.Payload, &env); err != nil { | |
| return false | |
| } |
🤖 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/mail/match.go` around lines 31 - 33, The handler currently does
json.Unmarshal(raw.Payload, &env) and returns true on error (fail-open), which
can route events to the wrong mailbox when params["mailbox"] is set; change the
logic in the mailbox filter to fail-closed by returning false when
json.Unmarshal returns an error (optionally logging the unmarshal error and the
offending raw.Payload and/or rejecting the event), i.e., in the block around
json.Unmarshal(raw.Payload, &env) switch the return value on err from true to
false so the event is rejected rather than passed through to per-mailbox
subscribers.
Comment on lines
+150
to
+156
| scope := m.SubscriptionID | ||
| if scope == "" { | ||
| scope = m.EventKey | ||
| } | ||
| lastForKey := true | ||
| if c.checkLastForKey != nil { | ||
| lastForKey = c.checkLastForKey(m.EventKey) | ||
| lastForKey = c.checkLastForKey(scope) |
There was a problem hiding this comment.
Use the connection’s registered subscription scope for PreShutdownCheck.
This path currently trusts m.SubscriptionID/m.EventKey, but the bus already knows the scope for this Conn. If a buggy or stale client sends the wrong identity here, checkLastForKey can answer for another subscription and either skip cleanup or unsubscribe the wrong mailbox stream. Prefer c.SubscriptionID() (optionally logging a mismatch) when computing scope.
Suggested fix
case *protocol.PreShutdownCheck:
- scope := m.SubscriptionID
- if scope == "" {
- scope = m.EventKey
- }
+ scope := c.SubscriptionID()
lastForKey := true
if c.checkLastForKey != nil {
lastForKey = c.checkLastForKey(scope)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| scope := m.SubscriptionID | |
| if scope == "" { | |
| scope = m.EventKey | |
| } | |
| lastForKey := true | |
| if c.checkLastForKey != nil { | |
| lastForKey = c.checkLastForKey(m.EventKey) | |
| lastForKey = c.checkLastForKey(scope) | |
| scope := c.SubscriptionID() | |
| lastForKey := true | |
| if c.checkLastForKey != nil { | |
| lastForKey = c.checkLastForKey(scope) | |
| } |
🤖 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/conn.go` around lines 150 - 156, In PreShutdownCheck (the
Conn method where scope is computed from m.SubscriptionID/m.EventKey), stop
trusting the incoming message identity and use the connection’s registered scope
by calling c.SubscriptionID() when computing scope for c.checkLastForKey; if
m.SubscriptionID or m.EventKey differ from c.SubscriptionID(), optionally emit a
debug/warn log about the mismatch before proceeding so checkLastForKey always
evaluates the conn’s true subscription scope.
Comment on lines
+35
to
+55
| func TestNormalizeParams_ErrorIsWrappedWithEventKey(t *testing.T) { | ||
| // We test the error wrapping pattern in isolation: same call site Run uses. | ||
| keyDef := &event.KeyDefinition{ | ||
| Key: "test.evt_normalize_fail", | ||
| NormalizeParams: func(_ context.Context, _ event.APIClient, _ map[string]string) error { | ||
| return errors.New("simulated normalize failure") | ||
| }, | ||
| } | ||
| err := keyDef.NormalizeParams(context.Background(), &fakeRT{}, map[string]string{}) | ||
| if err == nil { | ||
| t.Fatal("expected error from NormalizeParams") | ||
| } | ||
| // Run wraps with: fmt.Errorf("normalize params for %s: %w", EventKey, err) | ||
| wrapped := fmt.Errorf("normalize params for %s: %w", keyDef.Key, err) | ||
| if !strings.Contains(wrapped.Error(), "normalize params for test.evt_normalize_fail:") { | ||
| t.Errorf("wrap format wrong: %v", wrapped) | ||
| } | ||
| if !strings.Contains(wrapped.Error(), "simulated normalize failure") { | ||
| t.Errorf("underlying error not propagated: %v", wrapped) | ||
| } | ||
| } |
There was a problem hiding this comment.
These tests mirror the expected strings instead of verifying the implementation.
Line 48 and Line 104 rebuild the formatting inside the test, so both cases still pass if the production code stops wrapping/logging those messages correctly. Please drive the real branch or extract the formatter/wrapper into a helper and assert that helper directly.
Also applies to: 99-109
🤖 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/consume_test.go` around lines 35 - 55, The test
currently reconstructs the wrapper string with fmt.Errorf instead of exercising
the real wrapping logic; update TestNormalizeParams_ErrorIsWrappedWithEventKey
to either (A) invoke the actual code path that performs the wrap (call the
production wrapper used around keyDef.NormalizeParams — the same function Run
uses) so the returned err is the wrapped error coming from that logic, or (B)
extract the wrapping/formatting into a helper (e.g., a new helper used by
Run/NormalizeParams wrapper) and assert that helper directly; then assert on the
returned error from that real wrapper (use strings.Contains or strings.HasPrefix
on err.Error() for the prefix "normalize params for "+keyDef.Key and use
errors.Is/unwrap to verify the underlying "simulated normalize failure") instead
of building a new fmt.Errorf in the test.
Comment on lines
+61
to
+67
| // Normalize params (resolve aliases like "me" -> real email) before fingerprint | ||
| // compute, PreConsume, Match, Process. Must happen BEFORE doHello so the | ||
| // SubscriptionID we send to bus reflects canonical values. | ||
| if keyDef.NormalizeParams != nil { | ||
| if err := keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params); err != nil { | ||
| return fmt.Errorf("normalize params for %s: %w", opts.EventKey, err) | ||
| } |
There was a problem hiding this comment.
Move the timeout context above NormalizeParams.
NormalizeParams is documented to allow OAPI calls, but Line 73 installs opts.Timeout only after normalization runs. A slow alias/lookup call here can block startup indefinitely even when the user passed --timeout.
Suggested fix
- // Normalize params (resolve aliases like "me" -> real email) before fingerprint
- // compute, PreConsume, Match, Process. Must happen BEFORE doHello so the
- // SubscriptionID we send to bus reflects canonical values.
- if keyDef.NormalizeParams != nil {
- if err := keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params); err != nil {
- return fmt.Errorf("normalize params for %s: %w", opts.EventKey, err)
- }
- }
-
- // Compute subscription identity from normalized params + SubscriptionKey flags.
- subscriptionID := ComputeSubscriptionID(keyDef, opts.Params)
-
if opts.Timeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, opts.Timeout)
defer cancel()
}
+
+ // Normalize params (resolve aliases like "me" -> real email) before fingerprint
+ // compute, PreConsume, Match, Process. Must happen BEFORE doHello so the
+ // SubscriptionID we send to bus reflects canonical values.
+ if keyDef.NormalizeParams != nil {
+ if err := keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params); err != nil {
+ return fmt.Errorf("normalize params for %s: %w", opts.EventKey, err)
+ }
+ }
+
+ // Compute subscription identity from normalized params + SubscriptionKey flags.
+ subscriptionID := ComputeSubscriptionID(keyDef, opts.Params)Also applies to: 73-77
🤖 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/consume.go` around lines 61 - 67, Move the
timeout-wrapping of the request context so it is applied before calling
keyDef.NormalizeParams to ensure NormalizeParams runs under opts.Timeout;
specifically, in the Consume flow wrap the incoming ctx with the timeout (the
existing opts.Timeout installation code) before invoking
keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params) and ensure the deferred
cancel is still called after NormalizeParams returns; adjust any references to
ctx passed to NormalizeParams, doHello, PreConsume, Match, and Process so they
all use the timeout-wrapped context.
Comment on lines
+117
to
+122
| On graceful exit (`--max-events`/`--timeout` reached, or SIGTERM/stdin EOF), the last consumer for a `SubscriptionID` runs cleanup: a POST to `unsubscribe`. Two stderr outcomes: | ||
|
|
||
| - Success: `[event] cleanup done.` | ||
| - Failure: `WARN: cleanup failed: <reason> (server-side subscribe is idempotent — residual record will be overwritten on next subscribe)` | ||
|
|
||
| The Feishu server-side subscribe record is `(app, user, event_type)`-keyed and idempotent, so a failed unsubscribe leaks at most one stale record per `(app, user)` until the next `event consume` for that key — which silently overwrites it. Agents **MUST NOT** manually call the unsubscribe API as a recovery action: the server has no reference counting, so a stray unsubscribe will silently kill another co-living consumer's stream. |
There was a problem hiding this comment.
Cleanup keying here contradicts the per-mailbox subscription model.
This section says the server-side record is keyed only by (app, user, event_type), but the doc earlier describes mailbox as part of subscription identity and says different mailboxes get independent subscriptions. That makes the overwrite/leak explanation inaccurate for operators debugging cleanup. Please update this paragraph to describe the mailbox/resource dimension consistently.
🤖 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 `@skills/lark-event/references/lark-event-mail.md` around lines 117 - 122,
Update the paragraph describing server-side subscription keying to include the
mailbox/resource dimension so it matches earlier documentation: change the
described key from "(app, user, event_type)" to include "mailbox" (e.g., "(app,
user, mailbox, event_type)"), and adjust the explanations about overwrite/leak
behavior and the warning about manual unsubscribe so they refer to per-mailbox
subscriptions and show that failed unsubscribe leaks are per-mailbox (not global
per-user) and that manual unsubscribe can terminate a different consumer on the
same mailbox; ensure the text references the same "mailbox" terminology used
earlier in this document.
Comment on lines
+157
to
+162
| When the consumer exits gracefully and is the last for its subscription scope, the framework runs the cleanup (unsubscribe) callback. Two stderr outcomes: | ||
|
|
||
| - Success: `[event] cleanup done.` | ||
| - Failure: `WARN: cleanup failed: <reason> (server-side subscribe is idempotent — residual record will be overwritten on next subscribe)` | ||
|
|
||
| **Important**: do NOT manually call unsubscribe APIs as a recovery action. Subscriptions are reference-counted implicitly by Lark's server (one record per `(app, user, event_type)`), so a stray unsubscribe will silently affect other co-living consumers. The WARN is informational; the next consumer's subscribe call self-heals. |
There was a problem hiding this comment.
Fix the server-side identity wording here.
This paragraph now conflicts with the per-mailbox model introduced above. Saying there is one record per (app, user, event_type) implies mailbox/resource qualifiers are ignored, and "reference-counted" is stronger than the behavior described elsewhere in this PR. Please rephrase this in terms of the subscription scope / SubscriptionID so readers do not infer that different mailbox subscriptions share one server record.
Suggested wording
-**Important**: do NOT manually call unsubscribe APIs as a recovery action. Subscriptions are reference-counted implicitly by Lark's server (one record per `(app, user, event_type)`), so a stray unsubscribe will silently affect other co-living consumers. The WARN is informational; the next consumer's subscribe call self-heals.
+**Important**: do NOT manually call unsubscribe APIs as a recovery action. Lark keeps one server-side subscription record per subscription scope (the same scope this CLI identifies via `SubscriptionID`; for mail that includes the mailbox/resource qualifier). A stray unsubscribe can therefore tear down another co-living consumer in that same scope. The WARN is informational; the next consumer's subscribe call self-heals.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| When the consumer exits gracefully and is the last for its subscription scope, the framework runs the cleanup (unsubscribe) callback. Two stderr outcomes: | |
| - Success: `[event] cleanup done.` | |
| - Failure: `WARN: cleanup failed: <reason> (server-side subscribe is idempotent — residual record will be overwritten on next subscribe)` | |
| **Important**: do NOT manually call unsubscribe APIs as a recovery action. Subscriptions are reference-counted implicitly by Lark's server (one record per `(app, user, event_type)`), so a stray unsubscribe will silently affect other co-living consumers. The WARN is informational; the next consumer's subscribe call self-heals. | |
| When the consumer exits gracefully and is the last for its subscription scope, the framework runs the cleanup (unsubscribe) callback. Two stderr outcomes: | |
| - Success: `[event] cleanup done.` | |
| - Failure: `WARN: cleanup failed: <reason> (server-side subscribe is idempotent — residual record will be overwritten on next subscribe)` | |
| **Important**: do NOT manually call unsubscribe APIs as a recovery action. Lark keeps one server-side subscription record per subscription scope (the same scope this CLI identifies via `SubscriptionID`; for mail that includes the mailbox/resource qualifier). A stray unsubscribe can therefore tear down another co-living consumer in that same scope. The WARN is informational; the next consumer's subscribe call self-heals. |
🤖 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 `@skills/lark-event/SKILL.md` around lines 157 - 162, The current text
incorrectly implies a single server record per `(app, user, event_type)` and
uses "reference-counted"; update the wording to state that subscriptions are
tracked by subscription scope (the SubscriptionID / per-mailbox qualifier) so
different mailbox/resource subscriptions have distinct server records, and
replace "reference-counted" with a gentler description (e.g., server-side
subscriptions are keyed by SubscriptionID and are idempotent for the same
scope), then rephrase the WARN explanation to say the warning is informational
and the next subscribe for the same SubscriptionID self-heals rather than
implying global shared state.
Collaborator
Author
|
Superseded by a same-repo PR (branch pushed directly to larksuite/cli for native CI). The fork-based PR triggered the external-contributor workflow-approval gate; reopening as an in-repo PR so checks run automatically. Same commits, rebased onto latest main. |
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
1 participant
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
Mail (and any future per-resource EventKey such as drive comments or task updates) subscribes at a finer granularity than IM: its subscription identity is
(EventKey, mailbox), notEventKeyalone. The current bus keys dedup onEventKeyonly, so a secondevent consumeagainst the same key with a different mailbox is silently skipped (PreConsume never fires, the consumer hangs with no error). Cleanup failures are also swallowed. This PR lifts the framework from one-dimensional to two-dimensional subscription identity, adds a syncMatchfilter and aNormalizeParamshook, surfaces cleanup errors, and adapts the mail EventKey end-to-end as the proving ground. Raised in response to the review on #851.Changes
ParamDef.SubscriptionKey,KeyDefinition.NormalizeParams,KeyDefinition.Match; changePreConsumecleanup fromfunc()tofunc() errorininternal/event/types.goComputeSubscriptionID(sha256 + base64url, 16-char) ininternal/event/consume/fingerprint.goSubscriptionIDthrough the wire:Hello,PreShutdownCheck,ConsumerInfoininternal/event/protocol/messages.goSubscriptionID(was EventKey); addSubCount; preserveEventKeyCountas cross-subscription aggregate ininternal/event/bus/hub.go,conn.go,bus.goNormalizeParams+ComputeSubscriptionIDinto the consume run, plumbSubscriptionIDthroughcheckLastForKey, surface cleanup errors with an idempotency note ininternal/event/consume/{consume,handshake,loop,shutdown}.goSUB-KEYcolumn incmd/event/schema.goandSUBcolumn incmd/event/status.gomealias normalization (/user_mailboxes/me/profile), mailboxMatch, folders/labels filter + msg-format enrichmentProcessinevents/mail/{payload,normalize,match,process,register}.goskills/lark-event/; remove the mail-domain+watchskill surface so event subscription converges underlark-event(IM-style)SubscriptionIDfalls back to EventKey, so old consumer ↔ new daemon and new consumer ↔ old daemon both degrade to today's single-dimension behaviorTest Plan
make unit-testpassed (allinternal/event/...,events/...,cmd/event/...green with-race)lark-cli event consume mail.user_mailbox.event.message_received_v1 -p mailbox=me -p msg-format=metadata --max-events 1 --as user— normalize resolvedme, real mail event emitted with from/subject/snippet, cleanup ran cleanly;event schema mail.xxxshows the SUB-KEY column;event statusshows the SUB columnRelated Issues
N/A
Summary by CodeRabbit
Release Notes
New Features
mail.user_mailbox.event.message_received_v1) with mailbox selection, folder/label filtering, and configurable message formats.event statuswith new "SUB" column displaying subscription identifiers.event schemawith "SUB-KEY" column indicating subscription-identifying parameters.Documentation
Chores