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