feat: add Guard end-to-end demo

Author: Astoriel

README.md

@@ -126,6 +126,20 @@ Replay fixed DLQ events after updating the contract:
 go run ./cmd/trackboard-guard replay --config ../../examples/guard/guard.local.yaml --limit 100
 ```
 
+## End-to-End Demo
+
+Run Guard with a fake analytics destination and verify that a valid event is forwarded while an invalid event is blocked:
+
+```powershell
+.\scripts\demo-guard-e2e.ps1
+```
+
+Docker Compose can run the same runtime demo:
+
+```bash
+docker compose up --build guard fake-destination
+```
+
 ## Backend Install
 
 Use Python 3.12 and PostgreSQL. SQLite is not the serious backend path.
@@ -174,5 +188,6 @@ go test ./...
 - [GitHub Action](docs/github-action.md)
 - [Guard](docs/guard.md)
 - [Guard operations](docs/guard-operations.md)
+- [End-to-end demo](docs/demo.md)
 - [Product comparison](docs/comparison.md)
 - [API reference](docs/api-reference.md)

apps/guard/Dockerfile

@@ -0,0 +1,15 @@
+FROM golang:1.26.4-alpine AS build
+
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -o /out/trackboard-guard ./cmd/trackboard-guard
+
+FROM alpine:3.21
+
+RUN adduser -D -H trackboard
+USER trackboard
+COPY --from=build /out/trackboard-guard /usr/local/bin/trackboard-guard
+
+ENTRYPOINT ["trackboard-guard"]

apps/guard/internal/config/config.go

@@ -13,16 +13,18 @@ type Config struct {
 	ContractFile string
 	StoreFile    string
 	Destination  string
+	WorkerCount  int
 	Ready        bool
 	Mode         string
 }
 
 func Default() Config {
 	return Config{
-		HTTPAddr:  ":8080",
-		StoreFile: "guard.db",
-		Ready:     false,
-		Mode:      "block",
+		HTTPAddr:    ":8080",
+		StoreFile:   "guard.db",
+		WorkerCount: 1,
+		Ready:       false,
+		Mode:        "block",
 	}
 }
 
@@ -55,6 +57,12 @@ func Load(path string) (Config, error) {
 			cfg.StoreFile = value
 		case "destination":
 			cfg.Destination = value
+		case "worker_count":
+			workerCount, err := strconv.Atoi(value)
+			if err != nil {
+				return cfg, fmt.Errorf("invalid worker_count value: %w", err)
+			}
+			cfg.WorkerCount = workerCount
 		case "ready":
 			ready, err := strconv.ParseBool(value)
 			if err != nil {

apps/guard/internal/config/config_test.go

@@ -9,7 +9,7 @@ import (
 func TestLoadConfig(t *testing.T) {
 	dir := t.TempDir()
 	path := filepath.Join(dir, "guard.yaml")
-	err := os.WriteFile(path, []byte("http_addr: \":9090\"\ncontract_file: \"contract.json\"\nstore_file: \"guard.db\"\ndestination: \"http://example.test/track\"\nready: true\n"), 0o600)
+	err := os.WriteFile(path, []byte("http_addr: \":9090\"\ncontract_file: \"contract.json\"\nstore_file: \"guard.db\"\ndestination: \"http://example.test/track\"\nworker_count: 2\nready: true\n"), 0o600)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -31,6 +31,9 @@ func TestLoadConfig(t *testing.T) {
 	if cfg.Destination != "http://example.test/track" {
 		t.Fatalf("Destination = %q", cfg.Destination)
 	}
+	if cfg.WorkerCount != 2 {
+		t.Fatalf("WorkerCount = %d", cfg.WorkerCount)
+	}
 	if !cfg.Ready {
 		t.Fatal("Ready = false")
 	}

apps/guard/internal/server/server.go

@@ -6,21 +6,25 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/astoriel/trackboard/apps/guard/internal/config"
 	"github.com/astoriel/trackboard/apps/guard/internal/contract"
+	"github.com/astoriel/trackboard/apps/guard/internal/forwarder"
 	"github.com/astoriel/trackboard/apps/guard/internal/health"
 	"github.com/astoriel/trackboard/apps/guard/internal/ingest"
 	"github.com/astoriel/trackboard/apps/guard/internal/metrics"
 	"github.com/astoriel/trackboard/apps/guard/internal/policy"
+	"github.com/astoriel/trackboard/apps/guard/internal/runtime"
 	"github.com/astoriel/trackboard/apps/guard/internal/store"
 	"github.com/astoriel/trackboard/apps/guard/internal/validator"
 )
 
 type Server struct {
-	httpServer *http.Server
-	eventStore *store.Store
+	httpServer   *http.Server
+	eventStore   *store.Store
+	workerCancel context.CancelFunc
 }
 
 func New(cfg config.Config) *Server {
@@ -39,6 +43,16 @@ func New(cfg config.Config) *Server {
 			eventStore = opened
 		}
 	}
+	var workerCancel context.CancelFunc
+	if eventStore != nil && isHTTPDestination(cfg.Destination) {
+		workerCtx, cancel := context.WithCancel(context.Background())
+		workerCancel = cancel
+		go runtime.WorkerPool{
+			Store:       eventStore,
+			Sender:      forwarder.NewSegment(cfg.Destination),
+			WorkerCount: cfg.WorkerCount,
+		}.Run(workerCtx)
+	}
 	guardMetrics := metrics.New()
 	mux.HandleFunc("GET /health/live", healthHandler.Live)
 	mux.HandleFunc("GET /health/ready", healthHandler.Ready)
@@ -51,10 +65,15 @@ func New(cfg config.Config) *Server {
 			Handler:           mux,
 			ReadHeaderTimeout: 5 * time.Second,
 		},
-		eventStore: eventStore,
+		eventStore:   eventStore,
+		workerCancel: workerCancel,
 	}
 }
 
+func isHTTPDestination(destination string) bool {
+	return strings.HasPrefix(destination, "http://") || strings.HasPrefix(destination, "https://")
+}
+
 func trackHandler(cache *contract.Cache, eventStore *store.Store, destination string, mode policy.Mode, guardMetrics *metrics.Metrics) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if cache == nil {
@@ -162,6 +181,9 @@ func (s *Server) ListenAndServe() error {
 }
 
 func (s *Server) Shutdown(ctx context.Context) error {
+	if s.workerCancel != nil {
+		s.workerCancel()
+	}
 	err := s.httpServer.Shutdown(ctx)
 	if s.eventStore != nil {
 		if closeErr := s.eventStore.Close(); err == nil {

apps/guard/internal/server/server_test.go

@@ -7,7 +7,9 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/astoriel/trackboard/apps/guard/internal/config"
 	"github.com/astoriel/trackboard/apps/guard/internal/store"
@@ -61,6 +63,40 @@ func TestTrackEndpointQueuesValidEventsInOutbox(t *testing.T) {
 	assertDepths(t, storePath, 1, 0)
 }
 
+func TestTrackEndpointForwardsValidEventsToDestination(t *testing.T) {
+	contractPath := writeServerContract(t)
+	storePath := dbPath(t)
+	var received atomic.Int32
+	destination := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/track" {
+			t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path)
+		}
+		received.Add(1)
+		w.WriteHeader(http.StatusAccepted)
+	}))
+	defer destination.Close()
+	srv := New(config.Config{HTTPAddr: ":0", ContractFile: contractPath, StoreFile: storePath, Destination: destination.URL + "/track", Mode: "block", WorkerCount: 1})
+	defer shutdownServer(t, srv)
+	req := httptest.NewRequest(http.MethodPost, "/v1/track", strings.NewReader(`{
+		"event":"signup_completed",
+		"userId":"usr_123",
+		"messageId":"msg_forwarded",
+		"properties":{"user_id":"usr_123","signup_method":"google"}
+	}`))
+
+	srv.httpServer.Handler.ServeHTTP(httptest.NewRecorder(), req)
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if received.Load() == 1 {
+			assertDepths(t, storePath, 0, 0)
+			return
+		}
+		time.Sleep(25 * time.Millisecond)
+	}
+	t.Fatalf("destination received %d events", received.Load())
+}
+
 func TestMetricsEndpointReportsAcceptedAndBlockedEvents(t *testing.T) {
 	contractPath := writeServerContract(t)
 	srv := New(config.Config{HTTPAddr: ":0", ContractFile: contractPath, StoreFile: dbPath(t), Destination: "segment", Mode: "block"})

docker-compose.yml

@@ -64,5 +64,34 @@ services:
     depends_on:
       - api
 
+  fake-destination:
+    image: node:22-alpine
+    working_dir: /app
+    command: ["node", "server.js"]
+    volumes:
+      - ./examples/fake-destination:/app:ro
+    ports:
+      - "9000:9000"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://localhost:9000/health"]
+      interval: 5s
+      retries: 5
+
+  guard:
+    build:
+      context: ./apps/guard
+      dockerfile: Dockerfile
+    command: ["--config", "/config/guard.compose.yaml"]
+    volumes:
+      - ./examples/contracts:/contracts:ro
+      - ./examples/guard/guard.compose.yaml:/config/guard.compose.yaml:ro
+      - guarddata:/data
+    ports:
+      - "8080:8080"
+    depends_on:
+      fake-destination:
+        condition: service_healthy
+
 volumes:
   pgdata:
+  guarddata:

docs/demo.md

@@ -0,0 +1,61 @@
+# End-to-End Demo
+
+This demo proves the full runtime path:
+
+1. Guard loads a published Trackboard contract.
+2. A valid Segment-compatible event is accepted into the outbox.
+3. The Guard worker forwards the valid event to a fake analytics destination.
+4. An invalid event is blocked and recorded in Guard metrics.
+
+## Local Smoke Test
+
+```powershell
+.\scripts\demo-guard-e2e.ps1
+```
+
+Expected output:
+
+```text
+Guard E2E demo passed: valid event forwarded, invalid event blocked
+```
+
+The script starts:
+
+- `examples/fake-destination/server.js` on `localhost:9000`
+- `apps/guard` on `localhost:8080`
+
+It stops both processes when the check completes.
+
+## Docker Compose
+
+```bash
+docker compose up --build guard fake-destination
+```
+
+Then send a valid event:
+
+```bash
+curl -X POST http://localhost:8080/v1/track \
+  -H "Content-Type: application/json" \
+  -d @examples/events/signup.valid.json
+```
+
+Check that the fake destination received it:
+
+```bash
+curl http://localhost:9000/events
+```
+
+Send an invalid event:
+
+```bash
+curl -X POST http://localhost:8080/v1/track \
+  -H "Content-Type: application/json" \
+  -d @examples/events/signup.invalid.json
+```
+
+Check Guard metrics:
+
+```bash
+curl http://localhost:8080/metrics
+```

docs/guard-operations.md

@@ -15,6 +15,8 @@ ready: true
 mode: "block"
 ```
 
+When `destination` is an `http://` or `https://` URL, Guard starts a local forwarding worker. The worker leases due rows from the SQLite outbox and sends them to the configured destination.
+
 ## Metrics
 
 `GET /metrics` returns plain counters for:

docs/guard.md

@@ -10,7 +10,7 @@ Trackboard Guard is the runtime data plane. It validates Segment-compatible `/v1
 - Forward accepted events to an HTTP destination.
 - Store accepted events in a durable single-node SQLite outbox before forwarding.
 - Store rejected events in a single-node SQLite DLQ.
-- Retry failed forwarding attempts.
+- Run a local worker that leases outbox rows, forwards them, and retries failed delivery attempts.
 - Expose health and metrics endpoints.
 - Replay fixed DLQ events after the contract is updated.