1// Package eventbus implements the event bus adapter.
2package eventbus
3
4import (
5 "context"
6 "fmt"
7 "sync"
8 "time"
9
10 "github.com/bnema/zerowrap"
11 "github.com/google/uuid"
12 "go.opentelemetry.io/otel/attribute"
13 "go.opentelemetry.io/otel/metric"
14
15 "github.com/bnema/gordon/internal/adapters/out/telemetry"
16 "github.com/bnema/gordon/internal/boundaries/out"
17 "github.com/bnema/gordon/internal/domain"
18)
19
20// InMemory implements the EventBus interface using in-memory channels.
21type InMemory struct {
22 handlers []out.EventHandler
23 eventChan chan domain.Event
24 done chan struct{}
25 mu sync.RWMutex
26 ctx context.Context
27 cancel context.CancelFunc
28 bufferSize int
29 log zerowrap.Logger
30 metrics *telemetry.Metrics
31}
32
33// SetMetrics sets the telemetry metrics for the event bus.
34// Must be called before Start() to avoid data races on bus.metrics reads.
35func (bus *InMemory) SetMetrics(m *telemetry.Metrics) {
36 bus.mu.Lock()
37 bus.metrics = m
38 bus.mu.Unlock()
39}
40
41// NewInMemory creates a new in-memory event bus.
42func NewInMemory(bufferSize int, log zerowrap.Logger) *InMemory {
43 if bufferSize <= 0 {
44 bufferSize = 100
45 }
46
47 ctx, cancel := context.WithCancel(context.Background())
48
49 return &InMemory{
50 handlers: make([]out.EventHandler, 0),
51 eventChan: make(chan domain.Event, bufferSize),
52 done: make(chan struct{}),
53 ctx: ctx,
54 cancel: cancel,
55 bufferSize: bufferSize,
56 log: log,
57 }
58}
59
60// Publish publishes an event to the bus.
61func (bus *InMemory) Publish(eventType domain.EventType, payload any) error {
62 event := domain.Event{
63 ID: uuid.New().String(),
64 Type: eventType,
65 Timestamp: time.Now(),
66 Data: payload,
67 }
68
69 // Add specific fields for known event types
70 switch p := payload.(type) {
71 case domain.ImagePushedPayload:
72 event.ImageName = p.Name
73 event.Tag = p.Reference
74 }
75
76 select {
77 case bus.eventChan <- event:
78 bus.log.Debug().
79 Str(zerowrap.FieldLayer, "adapter").
80 Str(zerowrap.FieldAdapter, "eventbus").
81 Str("event_id", event.ID).
82 Str(zerowrap.FieldEvent, string(event.Type)).
83 Str("image_name", event.ImageName).
84 Str("tag", event.Tag).
85 Msg("event published")
86 return nil
87 case <-bus.ctx.Done():
88 return fmt.Errorf("event bus is stopped")
89 case <-time.After(5 * time.Second):
90 bus.log.Error().
91 Str(zerowrap.FieldLayer, "adapter").
92 Str(zerowrap.FieldAdapter, "eventbus").
93 Str("event_id", event.ID).
94 Str(zerowrap.FieldEvent, string(event.Type)).
95 Str("image_name", event.ImageName).
96 Str("tag", event.Tag).
97 Msg("event channel is full, dropping event after 5s timeout")
98
99 // Record dropped event metric
100 if bus.metrics != nil {
101 bus.metrics.EventsDropped.Add(context.Background(), 1, metric.WithAttributes(
102 attribute.String("event_type", string(event.Type)),
103 ))
104 }
105 return fmt.Errorf("event channel is full, dropping event %s", event.ID)
106 }
107}
108
109// Subscribe adds an event handler to the bus.
110func (bus *InMemory) Subscribe(handler out.EventHandler) error {
111 bus.mu.Lock()
112 defer bus.mu.Unlock()
113
114 bus.handlers = append(bus.handlers, handler)
115 bus.log.Debug().
116 Str(zerowrap.FieldLayer, "adapter").
117 Str(zerowrap.FieldAdapter, "eventbus").
118 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
119 Int("total_handlers", len(bus.handlers)).
120 Msg("event handler subscribed")
121
122 return nil
123}
124
125// Unsubscribe removes an event handler from the bus.
126func (bus *InMemory) Unsubscribe(handler out.EventHandler) error {
127 bus.mu.Lock()
128 defer bus.mu.Unlock()
129
130 for i, h := range bus.handlers {
131 if h == handler {
132 bus.handlers = append(bus.handlers[:i], bus.handlers[i+1:]...)
133 bus.log.Debug().
134 Str(zerowrap.FieldLayer, "adapter").
135 Str(zerowrap.FieldAdapter, "eventbus").
136 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
137 Int("total_handlers", len(bus.handlers)).
138 Msg("event handler unsubscribed")
139 return nil
140 }
141 }
142
143 return fmt.Errorf("handler not found")
144}
145
146// Start starts the event bus processing loop.
147func (bus *InMemory) Start() error {
148 bus.log.Info().
149 Str(zerowrap.FieldLayer, "adapter").
150 Str(zerowrap.FieldAdapter, "eventbus").
151 Int("buffer_size", bus.bufferSize).
152 Msg("starting event bus")
153
154 go bus.processEvents()
155 return nil
156}
157
158// Stop stops the event bus.
159func (bus *InMemory) Stop() error {
160 bus.log.Info().
161 Str(zerowrap.FieldLayer, "adapter").
162 Str(zerowrap.FieldAdapter, "eventbus").
163 Msg("stopping event bus")
164
165 bus.cancel()
166
167 select {
168 case <-bus.done:
169 bus.log.Info().
170 Str(zerowrap.FieldLayer, "adapter").
171 Str(zerowrap.FieldAdapter, "eventbus").
172 Msg("event bus stopped")
173 return nil
174 case <-time.After(5 * time.Second):
175 bus.log.Warn().
176 Str(zerowrap.FieldLayer, "adapter").
177 Str(zerowrap.FieldAdapter, "eventbus").
178 Msg("event bus stop timeout")
179 return fmt.Errorf("timeout waiting for event bus to stop")
180 }
181}
182
183func (bus *InMemory) processEvents() {
184 defer close(bus.done)
185
186 for {
187 select {
188 case event := <-bus.eventChan:
189 bus.handleEvent(event)
190 case <-bus.ctx.Done():
191 bus.log.Debug().
192 Str(zerowrap.FieldLayer, "adapter").
193 Str(zerowrap.FieldAdapter, "eventbus").
194 Msg("event bus processing stopped")
195 return
196 }
197 }
198}
199
200func (bus *InMemory) handleEvent(event domain.Event) {
201 bus.mu.RLock()
202 handlers := make([]out.EventHandler, len(bus.handlers))
203 copy(handlers, bus.handlers)
204 bus.mu.RUnlock()
205
206 for _, handler := range handlers {
207 if handler.CanHandle(event.Type) {
208 h := handler // shadow loop variable for goroutine capture
209 start := time.Now()
210
211 ctx, cancel := context.WithTimeout(bus.ctx, 30*time.Second)
212
213 done := make(chan error, 1)
214 go func() {
215 done <- h.Handle(ctx, event)
216 }()
217
218 select {
219 case err := <-done:
220 cancel()
221 if err != nil {
222 bus.log.Error().
223 Str(zerowrap.FieldLayer, "adapter").
224 Str(zerowrap.FieldAdapter, "eventbus").
225 Err(err).
226 Str("event_id", event.ID).
227 Str(zerowrap.FieldEvent, string(event.Type)).
228 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
229 Msg("error handling event")
230 } else {
231 bus.log.Debug().
232 Str(zerowrap.FieldLayer, "adapter").
233 Str(zerowrap.FieldAdapter, "eventbus").
234 Str("event_id", event.ID).
235 Str(zerowrap.FieldEvent, string(event.Type)).
236 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
237 Dur(zerowrap.FieldDuration, time.Since(start)).
238 Msg("event handled successfully")
239
240 // Record processed event metric
241 if bus.metrics != nil {
242 bus.metrics.EventsProcessed.Add(context.Background(), 1, metric.WithAttributes(
243 attribute.String("event_type", string(event.Type)),
244 ))
245 }
246 }
247 case <-ctx.Done():
248 cancel()
249 bus.log.Warn().
250 Str(zerowrap.FieldLayer, "adapter").
251 Str(zerowrap.FieldAdapter, "eventbus").
252 Str("event_id", event.ID).
253 Str(zerowrap.FieldEvent, string(event.Type)).
254 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
255 Dur(zerowrap.FieldDuration, time.Since(start)).
256 Msg("handler timeout after 30s")
257 }
258 }
259 }
260}