feat: implement event bus subscription and dispatch pipeline

.gitignore

@@ -28,5 +28,6 @@ vendor/
 #test
 test_scripts/
 tests/mail/reports/
+internal/registry/meta_data.json
 
 /log/

internal/registry/loader.go

@@ -19,9 +19,6 @@ import (
 //go:embed scope_priorities.json scope_overrides.json
 var registryFS embed.FS
 
-// embeddedMetaJSON is set by loader_embedded.go when meta_data.json is compiled in.
-var embeddedMetaJSON []byte
-
 var (
 	mergedServices    = make(map[string]map[string]interface{}) // project name → parsed spec
 	mergedProjectList []string                                  // sorted project names
@@ -43,7 +40,7 @@ func Init() {
 func InitWithBrand(brand core.LarkBrand) {
 	initOnce.Do(func() {
 		configuredBrand = brand
-		// 1. Load embedded meta_data.json as baseline (no-op if not compiled in)
+		// 1. Load embedded default registry baseline
 		loadEmbeddedIntoMerged()
 		// 2. Remote overlay
 		if remoteEnabled() && cacheWritable() {
@@ -69,8 +66,8 @@ func InitWithBrand(brand core.LarkBrand) {
 	})
 }
 
-// loadEmbeddedIntoMerged parses the embedded meta_data.json and populates
-// mergedServices. No-op if meta_data.json is not compiled in.
+// loadEmbeddedIntoMerged parses the embedded default registry JSON and populates
+// mergedServices. No-op if no embedded baseline is compiled in.
 func loadEmbeddedIntoMerged() {
 	if len(embeddedMetaJSON) == 0 {
 		return

internal/registry/loader_embedded.go

@@ -3,18 +3,7 @@
 
 package registry
 
-import "embed"
-
-//go:embed meta_data*.json
-var metaFS embed.FS
+import _ "embed"
 
 //go:embed meta_data_default.json
-var embeddedMetaDataDefaultJSON []byte
-
-func init() {
-	if data, err := metaFS.ReadFile("meta_data.json"); err == nil && len(data) > 0 {
-		embeddedMetaJSON = data
-	} else {
-		embeddedMetaJSON = embeddedMetaDataDefaultJSON
-	}
-}
+var embeddedMetaJSON []byte

internal/registry/meta_data_default.json

@@ -1 +1 @@
-{"version":"0.0.0","services":[]}
+{"version":"0.0.0","services":[{"name":"calendar","version":"v1","title":"calendar API","servicePath":"/open-apis/calendar/v1","resources":{"calendars":{"methods":{"get":{"id":"calendar.calendars.get","httpMethod":"GET","accessTokens":["user"],"scopes":["calendar:calendar:readonly","calendar:calendar:read"]},"create":{"id":"calendar.calendars.create","httpMethod":"POST","accessTokens":["user"],"scopes":["calendar:calendar:create"]}}},"events":{"methods":{"get":{"id":"calendar.events.get","httpMethod":"GET","accessTokens":["user"],"scopes":["calendar:calendar.event:read"]},"create":{"id":"calendar.events.create","httpMethod":"POST","accessTokens":["user"],"scopes":["calendar:calendar.event:create"]}}}}}]}

internal/registry/remote_test.go

@@ -29,7 +29,7 @@ func resetInit() {
 	testMetaURL = ""
 }
 
-// hasEmbeddedData returns true if meta_data.json is compiled in.
+// hasEmbeddedData returns true if the default embedded registry baseline is compiled in.
 func hasEmbeddedData() bool {
 	return len(embeddedMetaJSON) > 0
 }
@@ -77,7 +77,7 @@ func testEnvelopeNotModifiedJSON() []byte {
 
 func TestColdStart_UsesEmbedded(t *testing.T) {
 	if !hasEmbeddedData() {
-		t.Skip("no embedded from_meta data")
+		t.Skip("no embedded default registry baseline")
 	}
 	resetInit()
 	tmp := t.TempDir()

shortcuts/event/deduper.go

@@ -0,0 +1,34 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"sync"
+	"time"
+)
+
+// Deduper suppresses repeated keys seen within a TTL window.
+type Deduper struct {
+	ttl  time.Duration
+	seen sync.Map // key -> time.Time
+}
+
+// NewDeduper creates a deduper with the provided TTL.
+func NewDeduper(ttl time.Duration) *Deduper {
+	return &Deduper{ttl: ttl}
+}
+
+// Seen reports whether key has already been seen within ttl and records now.
+func (d *Deduper) Seen(key string, now time.Time) bool {
+	if d == nil || key == "" || d.ttl <= 0 {
+		return false
+	}
+	if v, loaded := d.seen.LoadOrStore(key, now); loaded {
+		if ts, ok := v.(time.Time); ok && now.Sub(ts) < d.ttl {
+			return true
+		}
+		d.seen.Store(key, now)
+	}
+	return false
+}

shortcuts/event/dispatcher.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import "context"
+
+// Dispatcher routes normalized events to registered handlers.
+type Dispatcher struct {
+	registry *HandlerRegistry
+}
+
+// NewDispatcher creates a dispatcher backed by the provided registry.
+func NewDispatcher(registry *HandlerRegistry) *Dispatcher {
+	if registry == nil {
+		registry = NewHandlerRegistry()
+	}
+	return &Dispatcher{registry: registry}
+}
+
+// Dispatch runs matching event handlers first, then matching domain handlers.
+// Fallback is only used when no direct handlers matched.
+func (d *Dispatcher) Dispatch(ctx context.Context, evt *Event) DispatchResult {
+	if d == nil || d.registry == nil || evt == nil {
+		return DispatchResult{}
+	}
+
+	matched := append([]EventHandler{}, d.registry.EventHandlers(evt.EventType)...)
+	matched = append(matched, d.registry.DomainHandlers(evt.Domain)...)
+	if len(matched) == 0 {
+		if fallback := d.registry.FallbackHandler(); fallback != nil {
+			matched = append(matched, fallback)
+		}
+	}
+
+	result := DispatchResult{Results: make([]DispatchRecord, 0, len(matched))}
+	for _, handler := range matched {
+		handlerResult := handler.Handle(ctx, evt)
+		result.Results = append(result.Results, DispatchRecord{
+			HandlerID: handler.ID(),
+			Status:    handlerResult.Status,
+			Reason:    handlerResult.Reason,
+			Err:       handlerResult.Err,
+			Output:    handlerResult.Output,
+		})
+	}
+	return result
+}

shortcuts/event/dispatcher_test.go

@@ -0,0 +1,210 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+)
+
+type testEventHandler struct {
+	id        string
+	eventType string
+	domain    string
+	result    HandlerResult
+	called    *[]string
+}
+
+func (h *testEventHandler) ID() string { return h.id }
+
+func (h *testEventHandler) EventType() string { return h.eventType }
+
+func (h *testEventHandler) Domain() string { return h.domain }
+
+func (h *testEventHandler) Handle(_ context.Context, _ *Event) HandlerResult {
+	if h.called != nil {
+		*h.called = append(*h.called, h.id)
+	}
+	return h.result
+}
+
+func TestDispatcher_EventHandlerThenDomainHandlerOrder(t *testing.T) {
+	registry := NewHandlerRegistry()
+	var calls []string
+
+	eventHandler := &testEventHandler{
+		id:        "event-handler",
+		eventType: "im.message.receive_v1",
+		result:    HandlerResult{Status: HandlerStatusHandled},
+		called:    &calls,
+	}
+	domainHandler := &testEventHandler{
+		id:     "domain-handler",
+		domain: "im",
+		result: HandlerResult{Status: HandlerStatusHandled},
+		called: &calls,
+	}
+
+	if err := registry.RegisterEventHandler(eventHandler); err != nil {
+		t.Fatalf("RegisterEventHandler() error = %v", err)
+	}
+	if err := registry.RegisterDomainHandler(domainHandler); err != nil {
+		t.Fatalf("RegisterDomainHandler() error = %v", err)
+	}
+
+	result := NewDispatcher(registry).Dispatch(context.Background(), &Event{
+		EventType: "im.message.receive_v1",
+		Domain:    "im",
+	})
+
+	if got, want := calls, []string{"event-handler", "domain-handler"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("call order = %v, want %v", got, want)
+	}
+	if len(result.Results) != 2 {
+		t.Fatalf("len(result.Results) = %d, want 2", len(result.Results))
+	}
+	if result.Results[0].HandlerID != "event-handler" || result.Results[1].HandlerID != "domain-handler" {
+		t.Fatalf("dispatch results = %+v", result.Results)
+	}
+}
+
+func TestNewBuiltinHandlerRegistry_RegistersRequiredIMHandlers(t *testing.T) {
+	registry := NewBuiltinHandlerRegistry()
+
+	if got, want := subscribedEventTypes, []string{
+		"im.message.receive_v1",
+		"im.message.message_read_v1",
+		"im.message.reaction.created_v1",
+		"im.message.reaction.deleted_v1",
+		"im.chat.member.bot.added_v1",
+		"im.chat.member.bot.deleted_v1",
+		"im.chat.member.user.added_v1",
+		"im.chat.member.user.withdrawn_v1",
+		"im.chat.member.user.deleted_v1",
+		"im.chat.updated_v1",
+		"im.chat.disbanded_v1",
+	}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("subscribedEventTypes = %v, want %v", got, want)
+	}
+
+	for _, eventType := range subscribedEventTypes {
+		handlers := registry.EventHandlers(eventType)
+		if len(handlers) != 1 {
+			t.Fatalf("EventHandlers(%q) len = %d, want 1", eventType, len(handlers))
+		}
+		if handlers[0].EventType() != eventType {
+			t.Fatalf("EventHandlers(%q)[0].EventType() = %q", eventType, handlers[0].EventType())
+		}
+		if handlers[0].Domain() != "im" {
+			t.Fatalf("EventHandlers(%q)[0].Domain() = %q, want im", eventType, handlers[0].Domain())
+		}
+	}
+
+	fallback := registry.FallbackHandler()
+	if fallback == nil {
+		t.Fatal("FallbackHandler() = nil")
+	}
+	if fallback.ID() != genericHandlerID {
+		t.Fatalf("fallback ID = %q, want %q", fallback.ID(), genericHandlerID)
+	}
+}
+
+func TestDispatcher_UsesFallbackWhenNoHandlersMatch(t *testing.T) {
+	registry := NewHandlerRegistry()
+	var calls []string
+	fallback := &testEventHandler{
+		id:     "fallback",
+		result: HandlerResult{Status: HandlerStatusSkipped, Reason: "no route"},
+		called: &calls,
+	}
+	registry.SetFallbackHandler(fallback)
+
+	result := NewDispatcher(registry).Dispatch(context.Background(), &Event{
+		EventType: "unknown.event",
+		Domain:    "unknown",
+	})
+
+	if got, want := calls, []string{"fallback"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("calls = %v, want %v", got, want)
+	}
+	if len(result.Results) != 1 {
+		t.Fatalf("len(result.Results) = %d, want 1", len(result.Results))
+	}
+	if result.Results[0].HandlerID != "fallback" {
+		t.Fatalf("fallback handler ID = %q, want fallback", result.Results[0].HandlerID)
+	}
+	if result.Results[0].Status != HandlerStatusSkipped {
+		t.Fatalf("fallback status = %q, want %q", result.Results[0].Status, HandlerStatusSkipped)
+	}
+}
+
+func TestHandlerRegistry_RejectsDuplicateHandlerID(t *testing.T) {
+	registry := NewHandlerRegistry()
+	if err := registry.RegisterEventHandler(&testEventHandler{
+		id:        "dup",
+		eventType: "im.message.receive_v1",
+		result:    HandlerResult{Status: HandlerStatusHandled},
+	}); err != nil {
+		t.Fatalf("first registration error = %v", err)
+	}
+
+	err := registry.RegisterDomainHandler(&testEventHandler{
+		id:     "dup",
+		domain: "im",
+		result: HandlerResult{Status: HandlerStatusHandled},
+	})
+	if err == nil {
+		t.Fatal("expected duplicate handler ID error")
+	}
+}
+
+func TestDispatcher_FailedHandlerDoesNotStopNextHandler(t *testing.T) {
+	registry := NewHandlerRegistry()
+	var calls []string
+	boom := errors.New("boom")
+
+	failed := &testEventHandler{
+		id:        "failed-handler",
+		eventType: "im.message.receive_v1",
+		result: HandlerResult{
+			Status: HandlerStatusFailed,
+			Reason: "failed",
+			Err:    boom,
+		},
+		called: &calls,
+	}
+	next := &testEventHandler{
+		id:        "next-handler",
+		eventType: "im.message.receive_v1",
+		result:    HandlerResult{Status: HandlerStatusHandled},
+		called:    &calls,
+	}
+
+	if err := registry.RegisterEventHandler(failed); err != nil {
+		t.Fatalf("RegisterEventHandler(failed) error = %v", err)
+	}
+	if err := registry.RegisterEventHandler(next); err != nil {
+		t.Fatalf("RegisterEventHandler(next) error = %v", err)
+	}
+
+	result := NewDispatcher(registry).Dispatch(context.Background(), &Event{EventType: "im.message.receive_v1"})
+
+	if got, want := calls, []string{"failed-handler", "next-handler"}; !reflect.DeepEqual(got, want) {
+		t.Fatalf("calls = %v, want %v", got, want)
+	}
+	if len(result.Results) != 2 {
+		t.Fatalf("len(result.Results) = %d, want 2", len(result.Results))
+	}
+	if result.Results[0].Status != HandlerStatusFailed {
+		t.Fatalf("first status = %q, want %q", result.Results[0].Status, HandlerStatusFailed)
+	}
+	if !errors.Is(result.Results[0].Err, boom) {
+		t.Fatalf("first error = %v, want boom", result.Results[0].Err)
+	}
+	if result.Results[1].Status != HandlerStatusHandled {
+		t.Fatalf("second status = %q, want %q", result.Results[1].Status, HandlerStatusHandled)
+	}
+}