jbrechtel
25881f3ff8
- Pub/sub broker with fan-out to SSE clients - POST /webhook — receives Gitea push events, optional HMAC verification - GET /events — SSE stream of push events - GET /health — health check endpoint - Zero external dependencies (stdlib only) - Docker-based build via goctl wrapper (golang:1.26-alpine) - systemd unit, README with deployment docs - Tests: broker pub/sub, slow subscriber drop, ref parsing, signature verification - 6.0M static binary
95 lines
2.5 KiB
Go
95 lines
2.5 KiB
Go
package main
|
|
|
|
import "log/slog"
|
|
|
|
// PushEvent is broadcast to all SSE subscribers when a webhook is received.
|
|
// It contains just enough information for converge clients to decide whether
|
|
// to pull.
|
|
type PushEvent struct {
|
|
FullName string `json:"full_name"` // e.g. "myorg/myrepo"
|
|
Branch string `json:"branch"` // e.g. "main"
|
|
Ref string `json:"ref"` // e.g. "refs/heads/main"
|
|
}
|
|
|
|
// Broker implements a pub/sub fan-out for PushEvents.
|
|
// Subscribe to get a channel that receives events; unsubscribe when done.
|
|
// Slow subscribers (channel full) are silently dropped for that event.
|
|
type Broker struct {
|
|
subscribe chan chan PushEvent
|
|
unsubscribe chan chan PushEvent
|
|
publish chan PushEvent
|
|
stop chan struct{}
|
|
subs map[chan PushEvent]struct{}
|
|
}
|
|
|
|
// NewBroker creates a ready-to-run Broker.
|
|
func NewBroker() *Broker {
|
|
return &Broker{
|
|
subscribe: make(chan chan PushEvent),
|
|
unsubscribe: make(chan chan PushEvent),
|
|
publish: make(chan PushEvent, 64),
|
|
stop: make(chan struct{}),
|
|
subs: make(map[chan PushEvent]struct{}),
|
|
}
|
|
}
|
|
|
|
// Run starts the broker event loop. Must be called in its own goroutine.
|
|
func (b *Broker) Run() {
|
|
for {
|
|
select {
|
|
case ch := <-b.subscribe:
|
|
b.subs[ch] = struct{}{}
|
|
slog.Debug("subscriber connected", "total", len(b.subs))
|
|
|
|
case ch := <-b.unsubscribe:
|
|
delete(b.subs, ch)
|
|
// Drain and close the channel so readers exit cleanly
|
|
close(ch)
|
|
for range ch {
|
|
}
|
|
slog.Debug("subscriber disconnected", "total", len(b.subs))
|
|
|
|
case evt := <-b.publish:
|
|
slog.Info("publishing event", "full_name", evt.FullName, "branch", evt.Branch)
|
|
for ch := range b.subs {
|
|
select {
|
|
case ch <- evt:
|
|
default:
|
|
// Subscriber too slow; drop the event for this subscriber
|
|
slog.Warn("dropping event for slow subscriber")
|
|
}
|
|
}
|
|
|
|
case <-b.stop:
|
|
for ch := range b.subs {
|
|
close(ch)
|
|
}
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Subscribe returns a buffered channel that receives PushEvents.
|
|
func (b *Broker) Subscribe() chan PushEvent {
|
|
// Small buffer to smooth out delivery; slow consumers are dropped by
|
|
// the broker's non-blocking publish anyway.
|
|
ch := make(chan PushEvent, 16)
|
|
b.subscribe <- ch
|
|
return ch
|
|
}
|
|
|
|
// Unsubscribe removes a subscriber channel.
|
|
func (b *Broker) Unsubscribe(ch chan PushEvent) {
|
|
b.unsubscribe <- ch
|
|
}
|
|
|
|
// Publish sends an event to all current subscribers. Never blocks.
|
|
func (b *Broker) Publish(evt PushEvent) {
|
|
b.publish <- evt
|
|
}
|
|
|
|
// Stop shuts down the broker.
|
|
func (b *Broker) Stop() {
|
|
close(b.stop)
|
|
}
|