feat: implement event bus subscription and dispatch pipeline

shortcuts/event/concurrency_test.go

@@ -0,0 +1,85 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+)
+
+func makeInboundEnvelopeWithEventID(eventID, eventType, eventJSON string) InboundEnvelope {
+	body := fmt.Sprintf(`{"schema":"2.0","header":{"event_id":"%s","event_type":"%s"},"event":%s}`, eventID, eventType, eventJSON)
+	return InboundEnvelope{
+		Source:     SourceWebSocket,
+		ReceivedAt: nowForTest(),
+		RawPayload: []byte(body),
+	}
+}
+
+func nowForTest() time.Time {
+	return time.Unix(1700000000, 0).UTC()
+}
+
+func TestOutputRouterWriteRecordConcurrent(t *testing.T) {
+	router := &outputRouter{
+		defaultDir: filepath.Join(t.TempDir(), "events"),
+		seq:        new(uint64),
+		writers:    map[string]*dirRecordWriter{},
+	}
+
+	const workers = 64
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			if err := router.WriteRecord("im.message.receive_v1", map[string]interface{}{
+				"event_type": "im.message.receive_v1",
+				"event_id":   fmt.Sprintf("evt-%03d", i),
+			}); err != nil {
+				t.Errorf("WriteRecord() error = %v", err)
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
+func TestPipelineConcurrentProcessCountsAllDispatches(t *testing.T) {
+	registry := NewHandlerRegistry()
+	if err := registry.RegisterEventHandler(handlerFuncWith{
+		id:        "counting-handler",
+		eventType: "im.message.receive_v1",
+		fn: func(_ context.Context, evt *Event) HandlerResult {
+			return HandlerResult{Status: HandlerStatusHandled, Output: map[string]interface{}{"event_id": evt.EventID}}
+		},
+	}); err != nil {
+		t.Fatalf("RegisterEventHandler() error = %v", err)
+	}
+
+	p := NewEventPipeline(registry, NewFilterChain(), PipelineConfig{Mode: TransformCompact}, io.Discard, io.Discard)
+
+	const workers = 64
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			p.Process(context.Background(), makeInboundEnvelopeWithEventID(
+				fmt.Sprintf("evt-%03d", i),
+				"im.message.receive_v1",
+				fmt.Sprintf(`{"message":{"message_id":"om_%03d"}}`, i),
+			))
+		}(i)
+	}
+	wg.Wait()
+
+	if got, want := p.EventCount(), int64(workers); got != want {
+		t.Fatalf("EventCount() = %d, want %d", got, want)
+	}
+}

shortcuts/event/deduper.go

@@ -0,0 +1,51 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const deduperCleanupInterval = 64
+
+// Deduper suppresses repeated keys seen within a TTL window.
+type Deduper struct {
+	ttl   time.Duration
+	seen  sync.Map // key -> time.Time
+	calls atomic.Uint64
+}
+
+// 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 d.calls.Add(1)%deduperCleanupInterval == 0 {
+		d.cleanup(now)
+	}
+	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
+}
+
+func (d *Deduper) cleanup(now time.Time) {
+	d.seen.Range(func(key, value any) bool {
+		ts, ok := value.(time.Time)
+		if ok && now.Sub(ts) >= d.ttl {
+			d.seen.Delete(key)
+		}
+		return true
+	})
+}

shortcuts/event/deduper_test.go

@@ -0,0 +1,44 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func deduperHasKey(d *Deduper, key string) bool {
+	if d == nil {
+		return false
+	}
+	found := false
+	d.seen.Range(func(k, _ any) bool {
+		if k == key {
+			found = true
+			return false
+		}
+		return true
+	})
+	return found
+}
+
+func TestDeduperEvictsExpiredKeysDuringSteadyState(t *testing.T) {
+	d := NewDeduper(time.Second)
+	now := time.Unix(100, 0).UTC()
+	if d.Seen("stale", now) {
+		t.Fatal("first observation of stale key should not dedupe")
+	}
+
+	later := now.Add(2 * time.Second)
+	for i := 0; i < 128; i++ {
+		if d.Seen(fmt.Sprintf("fresh-%03d", i), later) {
+			t.Fatalf("fresh key %d should not dedupe on first observation", i)
+		}
+	}
+
+	if deduperHasKey(d, "stale") {
+		t.Fatal("stale key should be evicted after periodic cleanup")
+	}
+}

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
+}