From 25881f3ff87a3dfa57e15f8b928a48352e3794b4 Mon Sep 17 00:00:00 2001 From: James Brechtel Date: Thu, 14 May 2026 13:54:26 -0400 Subject: [PATCH] =?UTF-8?q?converge-relay:=20phase=201=20=E2=80=94=20webho?= =?UTF-8?q?ok-to-SSE=20relay=20server=20in=20Go?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .gitignore | 2 + converge-relay/README.md | 89 ++++++++++++++++ converge-relay/broker.go | 94 +++++++++++++++++ converge-relay/broker_test.go | 140 ++++++++++++++++++++++++++ converge-relay/converge-relay.service | 18 ++++ converge-relay/go.mod | 3 + converge-relay/goctl | 20 ++++ converge-relay/main.go | 58 +++++++++++ converge-relay/scripts/build | 14 +++ converge-relay/scripts/run | 4 + converge-relay/scripts/test | 4 + converge-relay/sse.go | 59 +++++++++++ converge-relay/webhook.go | 96 ++++++++++++++++++ 13 files changed, 601 insertions(+) create mode 100644 converge-relay/README.md create mode 100644 converge-relay/broker.go create mode 100644 converge-relay/broker_test.go create mode 100644 converge-relay/converge-relay.service create mode 100644 converge-relay/go.mod create mode 100755 converge-relay/goctl create mode 100644 converge-relay/main.go create mode 100755 converge-relay/scripts/build create mode 100755 converge-relay/scripts/run create mode 100755 converge-relay/scripts/test create mode 100644 converge-relay/sse.go create mode 100644 converge-relay/webhook.go diff --git a/.gitignore b/.gitignore index 8e91cba..4ca7112 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,5 @@ dist/ dist-newstyle/ .claude/ .pi/ +converge-relay/.go-cache/ +converge-relay/build/ diff --git a/converge-relay/README.md b/converge-relay/README.md new file mode 100644 index 0000000..660b751 --- /dev/null +++ b/converge-relay/README.md @@ -0,0 +1,89 @@ +# converge-relay + +Webhook-to-SSE relay for [converge](https://github.com/jbrechtel/converge). Receives Gitea push webhooks and fans them out to connected converge clients via Server-Sent Events. + +## How it works + +``` +┌─────────────┐ POST /webhook ┌──────────────────┐ GET /events (SSE) ┌──────────────────┐ +│ Gitea │ ────────────────> │ converge-relay │ ──────────────────> │ converge (local) │ +│ Server │ (push event) │ (this server) │ event: push │ │ +└─────────────┘ └──────────────────┘ └──────────────────┘ +``` + +1. Configure Gitea to send push webhooks to `https://your-server:8080/webhook` +2. Run `converge-relay` on a publicly reachable server (or behind nginx) +3. Point your local `converge` at the relay's SSE endpoint with `--relay-url http://your-server:8080/events` + +## Build & run (Docker-only, no host Go tools required) + +```bash +# Build (formats, tests, compiles) → ./build/converge-relay +./scripts/build + +# Run tests only +./scripts/test + +# Run in development +./scripts/run --listen :8080 + +# Run go commands directly +./goctl go fmt ./... +./goctl go mod tidy +``` + +## Configuration + +| Flag | Env var | Default | Description | +|------|---------|---------|-------------| +| `--listen` | `CONVERGE_RELAY_LISTEN` | `:8080` | Address to listen on | +| `--secret` | `CONVERGE_RELAY_SECRET` | `""` | Shared secret for HMAC webhook verification | + +## Endpoints + +### `POST /webhook` + +Receives Gitea push webhooks. Validates `X-Gitea-Signature` or `X-Hub-Signature-256` against the configured secret. Extracts `repository.full_name` and branch name from `ref`. + +### `GET /events` + +SSE stream. Clients connect and receive push events as they arrive: + +``` +event: push +data: {"full_name":"myorg/myrepo","branch":"main","ref":"refs/heads/main"} +``` + +Automatic reconnection with `Last-Event-ID` is supported. When the connection drops, clients reconnect with exponential backoff. + +### `GET /health` + +Returns `ok\n` — useful for load balancer health checks. + +## Deployment + +```bash +# Build +./scripts/build + +# Install +sudo cp build/converge-relay /usr/local/bin/ + +# Install and enable the systemd service +sudo cp converge-relay.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now converge-relay +``` + +For production, set the `CONVERGE_RELAY_SECRET` environment variable in the service file or drop-in. + +## Gitea webhook setup + +1. Go to your repository → Settings → Webhooks → Add Webhook → Gitea +2. Target URL: `https://your-server:8080/webhook` +3. Secret: same value as `CONVERGE_RELAY_SECRET` +4. Events: Push events only + +## Docker image + +Built via `golang:1.26-alpine`. No external Go dependencies — stdlib only. diff --git a/converge-relay/broker.go b/converge-relay/broker.go new file mode 100644 index 0000000..627681b --- /dev/null +++ b/converge-relay/broker.go @@ -0,0 +1,94 @@ +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) +} diff --git a/converge-relay/broker_test.go b/converge-relay/broker_test.go new file mode 100644 index 0000000..f55525e --- /dev/null +++ b/converge-relay/broker_test.go @@ -0,0 +1,140 @@ +package main + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "net/http" + "testing" + "time" +) + +func TestBrokerSubscribePublishUnsubscribe(t *testing.T) { + broker := NewBroker() + go broker.Run() + defer broker.Stop() + + ch1 := broker.Subscribe() + ch2 := broker.Subscribe() + + evt := PushEvent{FullName: "org/repo", Branch: "main", Ref: "refs/heads/main"} + + // Publish + broker.Publish(evt) + + // Both subscribers should receive + select { + case got := <-ch1: + if got != evt { + t.Errorf("ch1: got %+v, want %+v", got, evt) + } + case <-time.After(time.Second): + t.Fatal("ch1: timed out waiting for event") + } + + select { + case got := <-ch2: + if got != evt { + t.Errorf("ch2: got %+v, want %+v", got, evt) + } + case <-time.After(time.Second): + t.Fatal("ch2: timed out waiting for event") + } + + // Unsubscribe ch2, publish again + broker.Unsubscribe(ch2) + // Drain ch2 (Unsubscribe closes it) + for range ch2 { + } + + evt2 := PushEvent{FullName: "org/repo2", Branch: "dev", Ref: "refs/heads/dev"} + broker.Publish(evt2) + + // ch1 gets it + select { + case got := <-ch1: + if got != evt2 { + t.Errorf("ch1 second: got %+v, want %+v", got, evt2) + } + case <-time.After(time.Second): + t.Fatal("ch1: timed out waiting for second event") + } + + // ch2 is closed, should not receive + _, open := <-ch2 + if open { + t.Error("ch2 should be closed after unsubscribe") + } +} + +func TestBrokerSlowSubscriberDropped(t *testing.T) { + broker := NewBroker() + go broker.Run() + defer broker.Stop() + + // Channel with buffer 0 — publish will always fail to send + ch := make(chan PushEvent) + broker.subs[ch] = struct{}{} // bypass Subscribe for manual control + + evt := PushEvent{FullName: "org/repo", Branch: "main"} + broker.Publish(evt) + + // Give it time to process + time.Sleep(50 * time.Millisecond) + + // Should not have blocked the broker; Publish returned fine + // and we didn't deadlock. + select { + case <-ch: + t.Error("slow subscriber should have been dropped") + default: + // expected: no event on unbuffered channel + } + // ch is closed by broker.Stop() via defer +} + +func TestRefToBranch(t *testing.T) { + tests := []struct { + ref string + want string + }{ + {"refs/heads/main", "main"}, + {"refs/heads/feature/x", "feature/x"}, + {"refs/tags/v1.0", ""}, + {"refs/heads/", ""}, + {"main", ""}, + } + for _, tt := range tests { + got := refToBranch(tt.ref) + if got != tt.want { + t.Errorf("refToBranch(%q) = %q, want %q", tt.ref, got, tt.want) + } + } +} + +func TestVerifySignature(t *testing.T) { + secret := "test-secret" + body := []byte(`{"ref":"refs/heads/main"}`) + + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + validSig := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + + headers := http.Header{} + headers.Set("X-Gitea-Signature", validSig) + + if !verifySignature(body, secret, headers) { + t.Error("valid signature should pass") + } + + headers.Set("X-Gitea-Signature", "sha256=bad") + if verifySignature(body, secret, headers) { + t.Error("bad signature should fail") + } + + headers.Del("X-Gitea-Signature") + headers.Set("X-Hub-Signature-256", validSig) + if !verifySignature(body, secret, headers) { + t.Error("X-Hub-Signature-256 with valid signature should pass") + } +} diff --git a/converge-relay/converge-relay.service b/converge-relay/converge-relay.service new file mode 100644 index 0000000..309d1da --- /dev/null +++ b/converge-relay/converge-relay.service @@ -0,0 +1,18 @@ +[Unit] +Description=Converge relay - webhook-to-SSE relay server +Documentation=https://github.com/jbrechtel/converge +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=%h/.local/bin/converge-relay --listen :8080 +Restart=on-failure +RestartSec=5 + +# Set a shared secret for webhook HMAC verification (optional but recommended). +# Generate with: openssl rand -hex 32 +# Environment=CONVERGE_RELAY_SECRET=your-secret-here + +[Install] +WantedBy=default.target diff --git a/converge-relay/go.mod b/converge-relay/go.mod new file mode 100644 index 0000000..0ef55fa --- /dev/null +++ b/converge-relay/go.mod @@ -0,0 +1,3 @@ +module github.com/jbrechtel/converge-relay + +go 1.26 diff --git a/converge-relay/goctl b/converge-relay/goctl new file mode 100755 index 0000000..c95e184 --- /dev/null +++ b/converge-relay/goctl @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Thin wrapper to run Go tooling inside a golang:1.26-alpine Docker container. +# Usage: ./goctl go build ./..., ./goctl go test ./..., ./goctl go fmt ./... +set -euo pipefail + +IMAGE="${CONVERGE_GO_IMAGE:-golang:1.26-alpine}" +PROJECT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +GO_CACHE="${PROJECT_DIR}/.go-cache" +mkdir -p "${GO_CACHE}/mod" "${GO_CACHE}/build" + +exec docker run --rm -i $([ -t 0 ] && printf -- -t) \ + -v "${PROJECT_DIR}:/work" \ + -v "${GO_CACHE}/mod:/go/pkg/mod" \ + -v "${GO_CACHE}/build:/root/.cache/go-build" \ + -e GOMODCACHE=/go/pkg/mod \ + -e GOCACHE=/root/.cache/go-build \ + -w /work \ + "${IMAGE}" \ + "$@" diff --git a/converge-relay/main.go b/converge-relay/main.go new file mode 100644 index 0000000..7fca9ec --- /dev/null +++ b/converge-relay/main.go @@ -0,0 +1,58 @@ +package main + +import ( + "context" + "errors" + "flag" + "log/slog" + "net/http" + "os" + "os/signal" +) + +func main() { + listen := flag.String("listen", envDefault("CONVERGE_RELAY_LISTEN", ":8080"), "address to listen on") + secret := flag.String("secret", envDefault("CONVERGE_RELAY_SECRET", ""), "shared secret for Gitea webhook HMAC verification") + flag.Parse() + + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})) + slog.SetDefault(logger) + + broker := NewBroker() + go broker.Run() + + mux := http.NewServeMux() + mux.HandleFunc("POST /webhook", webhookHandler(broker, *secret)) + mux.HandleFunc("GET /events", sseHandler(broker)) + mux.HandleFunc("GET /health", healthHandler) + + srv := &http.Server{Addr: *listen, Handler: mux} + + // Graceful shutdown + go func() { + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt) + <-sig + slog.Info("shutting down") + broker.Stop() + srv.Shutdown(context.Background()) + }() + + slog.Info("converge-relay starting", "listen", *listen) + if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) { + slog.Error("server error", "error", err) + os.Exit(1) + } +} + +func envDefault(key, fallback string) string { + if v, ok := os.LookupEnv(key); ok { + return v + } + return fallback +} + +func healthHandler(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain") + w.Write([]byte("ok\n")) +} diff --git a/converge-relay/scripts/build b/converge-relay/scripts/build new file mode 100755 index 0000000..dff017c --- /dev/null +++ b/converge-relay/scripts/build @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "${BASH_SOURCE[0]}")/.." + +echo "Formatting..." +./goctl go fmt ./... + +echo "Running tests..." +./goctl go test ./... + +echo "Building..." +./goctl go build -ldflags="-s -w" -o /work/build/converge-relay . + +echo "Done: $(ls -lh build/converge-relay | awk '{print $5}')" diff --git a/converge-relay/scripts/run b/converge-relay/scripts/run new file mode 100755 index 0000000..aa0c6a7 --- /dev/null +++ b/converge-relay/scripts/run @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "${BASH_SOURCE[0]}")/.." +./goctl go run . "$@" diff --git a/converge-relay/scripts/test b/converge-relay/scripts/test new file mode 100755 index 0000000..4e91568 --- /dev/null +++ b/converge-relay/scripts/test @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "${BASH_SOURCE[0]}")/.." +./goctl go test -v ./... diff --git a/converge-relay/sse.go b/converge-relay/sse.go new file mode 100644 index 0000000..1b5b2dc --- /dev/null +++ b/converge-relay/sse.go @@ -0,0 +1,59 @@ +package main + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" +) + +func sseHandler(broker *Broker) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming not supported", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") // disable nginx buffering + + ch := broker.Subscribe() + defer broker.Unsubscribe(ch) + + slog.Debug("SSE client connected") + + ctx := r.Context() + for { + select { + case <-ctx.Done(): + slog.Debug("SSE client disconnected") + return + + case evt, ok := <-ch: + if !ok { + // Broker shut down + return + } + writeSSEEvent(w, flusher, "push", evt) + } + } + } +} + +func writeSSEEvent(w http.ResponseWriter, flusher http.Flusher, eventType string, data any) { + jsonData, err := json.Marshal(data) + if err != nil { + slog.Warn("failed to marshal SSE event", "error", err) + return + } + + _, err = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, jsonData) + if err != nil { + slog.Debug("failed to write SSE event, client likely disconnected", "error", err) + return + } + flusher.Flush() +} diff --git a/converge-relay/webhook.go b/converge-relay/webhook.go new file mode 100644 index 0000000..8867967 --- /dev/null +++ b/converge-relay/webhook.go @@ -0,0 +1,96 @@ +package main + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "log/slog" + "net/http" + "strings" +) + +// giteaPushPayload mirrors the relevant fields from Gitea's push webhook JSON. +type giteaPushPayload struct { + Ref string `json:"ref"` + Repository struct { + FullName string `json:"full_name"` + } `json:"repository"` +} + +func webhookHandler(broker *Broker, secret string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1 MB limit + if err != nil { + slog.Warn("failed to read webhook body", "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + // Verify signature if a secret is configured + if secret != "" && !verifySignature(body, secret, r.Header) { + slog.Warn("webhook signature verification failed") + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + // Only handle push events + eventType := r.Header.Get("X-Gitea-Event") + if eventType != "" && eventType != "push" { + slog.Debug("ignoring non-push event", "type", eventType) + w.WriteHeader(http.StatusOK) + return + } + + var payload giteaPushPayload + if err := json.Unmarshal(body, &payload); err != nil { + slog.Warn("failed to parse webhook payload", "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + branch := refToBranch(payload.Ref) + if branch == "" { + slog.Debug("ignoring non-branch ref", "ref", payload.Ref) + w.WriteHeader(http.StatusOK) + return + } + + broker.Publish(PushEvent{ + FullName: payload.Repository.FullName, + Branch: branch, + Ref: payload.Ref, + }) + w.WriteHeader(http.StatusOK) + } +} + +// verifySignature checks the request body against the HMAC-SHA256 signature +// in X-Gitea-Signature or X-Hub-Signature-256 headers. +func verifySignature(body []byte, secret string, headers http.Header) bool { + sig := headers.Get("X-Gitea-Signature") + if sig == "" { + sig = headers.Get("X-Hub-Signature-256") + } + if sig == "" { + return false + } + + // Strip "sha256=" prefix if present + sig = strings.TrimPrefix(sig, "sha256=") + + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + expected := hex.EncodeToString(mac.Sum(nil)) + return hmac.Equal([]byte(sig), []byte(expected)) +} + +// refToBranch extracts the branch name from a git ref like "refs/heads/main". +func refToBranch(ref string) string { + const prefix = "refs/heads/" + if strings.HasPrefix(ref, prefix) { + return ref[len(prefix):] + } + return "" +}