converge-relay: phase 1 — webhook-to-SSE relay server in Go

- 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
2026-05-14 13:54:26 -04:00
parent 4f79212c45
commit 25881f3ff8
13 changed files with 601 additions and 0 deletions
@@ -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.
@@ -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)
+}
@@ -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")
+	}
+}
@@ -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
@@ -0,0 +1,3 @@
+module github.com/jbrechtel/converge-relay
+
+go 1.26
@@ -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}" \
+  "$@"
@@ -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"))
+}
@@ -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}')"
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "${BASH_SOURCE[0]}")/.."
+./goctl go run . "$@"
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "${BASH_SOURCE[0]}")/.."
+./goctl go test -v ./...
@@ -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()
+}
@@ -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 ""
+}