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