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 default:
78 return fmt.Errorf("event channel is full, dropping event %s", event.ID)
79 }
80}
81
82// Subscribe adds an event handler to the bus.
83func (bus *InMemory) Subscribe(handler out.EventHandler) error {
84 bus.mu.Lock()
85 defer bus.mu.Unlock()
86
87 bus.handlers = append(bus.handlers, handler)
88 bus.log.Debug().
89 Str(zerowrap.FieldLayer, "adapter").
90 Str(zerowrap.FieldAdapter, "eventbus").
91 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
92 Int("total_handlers", len(bus.handlers)).
93 Msg("event handler subscribed")
94
95 return nil
96}
97
98// Unsubscribe removes an event handler from the bus.
99func (bus *InMemory) Unsubscribe(handler out.EventHandler) error {
100 bus.mu.Lock()
101 defer bus.mu.Unlock()
102
103 for i, h := range bus.handlers {
104 if h == handler {
105 bus.handlers = append(bus.handlers[:i], bus.handlers[i+1:]...)
106 bus.log.Debug().
107 Str(zerowrap.FieldLayer, "adapter").
108 Str(zerowrap.FieldAdapter, "eventbus").
109 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
110 Int("total_handlers", len(bus.handlers)).
111 Msg("event handler unsubscribed")
112 return nil
113 }
114 }
115
116 return fmt.Errorf("handler not found")
117}
118
119// Start starts the event bus processing loop.
120func (bus *InMemory) Start() error {
121 bus.log.Info().
122 Str(zerowrap.FieldLayer, "adapter").
123 Str(zerowrap.FieldAdapter, "eventbus").
124 Int("buffer_size", bus.bufferSize).
125 Msg("starting event bus")
126
127 go bus.processEvents()
128 return nil
129}
130
131// Stop stops the event bus.
132func (bus *InMemory) Stop() error {
133 bus.log.Info().
134 Str(zerowrap.FieldLayer, "adapter").
135 Str(zerowrap.FieldAdapter, "eventbus").
136 Msg("stopping event bus")
137
138 bus.cancel()
139
140 select {
141 case <-bus.done:
142 bus.log.Info().
143 Str(zerowrap.FieldLayer, "adapter").
144 Str(zerowrap.FieldAdapter, "eventbus").
145 Msg("event bus stopped")
146 return nil
147 case <-time.After(5 * time.Second):
148 bus.log.Warn().
149 Str(zerowrap.FieldLayer, "adapter").
150 Str(zerowrap.FieldAdapter, "eventbus").
151 Msg("event bus stop timeout")
152 return fmt.Errorf("timeout waiting for event bus to stop")
153 }
154}
155
156func (bus *InMemory) processEvents() {
157 defer close(bus.done)
158
159 for {
160 select {
161 case event := <-bus.eventChan:
162 bus.handleEvent(event)
163 case <-bus.ctx.Done():
164 bus.log.Debug().
165 Str(zerowrap.FieldLayer, "adapter").
166 Str(zerowrap.FieldAdapter, "eventbus").
167 Msg("event bus processing stopped")
168 return
169 }
170 }
171}
172
173func (bus *InMemory) handleEvent(event domain.Event) {
174 bus.mu.RLock()
175 handlers := make([]out.EventHandler, len(bus.handlers))
176 copy(handlers, bus.handlers)
177 bus.mu.RUnlock()
178
179 for _, handler := range handlers {
180 if handler.CanHandle(event.Type) {
181 go func(h out.EventHandler) {
182 start := time.Now()
183
184 if err := h.Handle(event); err != nil {
185 bus.log.Error().
186 Str(zerowrap.FieldLayer, "adapter").
187 Str(zerowrap.FieldAdapter, "eventbus").
188 Err(err).
189 Str("event_id", event.ID).
190 Str(zerowrap.FieldEvent, string(event.Type)).
191 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
192 Msg("error handling event")
193 } else {
194 bus.log.Debug().
195 Str(zerowrap.FieldLayer, "adapter").
196 Str(zerowrap.FieldAdapter, "eventbus").
197 Str("event_id", event.ID).
198 Str(zerowrap.FieldEvent, string(event.Type)).
199 Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
200 Dur(zerowrap.FieldDuration, time.Since(start)).
201 Msg("event handled successfully")
202 }
203 }(handler)
204 }
205 }
206}