Consumer

.air.toml

@@ -8,7 +8,7 @@ tmp_dir = "tmp"
 [build]
   args_bin = []
   bin = "./tmp/main"
-  cmd = "go build -o ./tmp/main ."
+  cmd = "go build -o ./tmp/main ./cmd/processor"
   delay = 1000
   entrypoint = ["./tmp/main"]
   exclude_dir = ["assets", "tmp", "vendor", "testdata", ".git"]

.env.example

@@ -1 +1,10 @@
-PORT=3000
+PROCESSING_PORT=:3000
+REDPANDA_BROKERS=localhost:19092
+REDPANDA_SASL_USER=superuser
+REDPANDA_SASL_PASSWORD=12345678
+# SCRAM-SHA-256 (default) or SCRAM-SHA-512
+REDPANDA_SASL_MECHANISM=SCRAM-SHA-256
+REDIS_URL=redis://localhost:6379
+LOG_LEVEL=info
+OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4318
+CONSUMER_GROUP=event-processor

.github/workflows/ci.yml

@@ -0,0 +1,41 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+
+      - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      - name: go vet
+        run: go vet ./...
+
+      - name: go test
+        run: go test ./...
+
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+
+      - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20 # v9.2.0
+        with:
+          version: v2.11.3

.golangci.yml

@@ -0,0 +1,19 @@
+version: "2"
+
+linters:
+  default: none
+  enable:
+    - errcheck
+    - govet
+    - staticcheck
+  settings:
+    errcheck:
+      # Report errors from type assertions (e.g. v, ok := x.(T) when ok is ignored)
+      check-type-assertions: true
+      # Report when errors are assigned to the blank identifier
+      check-blank: true
+
+issues:
+  # Don't limit the number of issues per linter
+  max-issues-per-linter: 0
+  max-same-issues: 0

Dockerfile

@@ -5,25 +5,28 @@ WORKDIR /app
 COPY go.mod go.sum ./
 RUN go mod download
 
-COPY main.go .
+COPY cmd/ ./cmd/
+COPY internal/ ./internal/
 
 RUN CGO_ENABLED=0 GOOS=linux go build \
-    -ldflags="-s -w" \
-    -o app \
-    .
+  -ldflags="-s -w" \
+  -o bin/processor \
+  ./cmd/processor
 
-FROM alpine:latest
+FROM alpine:3.21.6
 
 RUN adduser -D -g '' appuser
 
 WORKDIR /app
 
-COPY --from=builder /app/processing .
+COPY --from=builder /app/bin/processor/ /bin/processor
 
 RUN chown -R appuser:appuser /app
 
 USER appuser
 
 EXPOSE 3000
 
-CMD ["./app"]
+HEALTHCHECK --interval=10s --timeout=3s CMD wget -qO- http://127.0.0.1:3000/healthz || exit 1
+
+CMD ["/bin/processor"]

cmd/processor/main.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/Tanilytics/processing/internal/config"
+	"github.com/Tanilytics/processing/internal/consumer"
+	"github.com/Tanilytics/processing/internal/observability"
+	"github.com/Tanilytics/processing/internal/server"
+)
+
+func main() {
+	// 1. Load configuration
+	cfg := config.LoadProcessorConfig()
+
+	// 2. Init structured logger
+	logger := observability.NewLogger("processor-service", cfg.LogLevel)
+	logger.Info("starting processor service",
+		"port", cfg.Port,
+		"brokers", cfg.RedpandaBrokers,
+		"redis_url", cfg.RedisURL,
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// 3. Init OTel Tracer
+	shutdownTracer, err := observability.InitTracer(ctx, "processor-service", cfg.OTelEndpoint)
+	if err != nil {
+		logger.Error("failed to initialize tracer", "error", err)
+	} else {
+		defer shutdownTracer()
+	}
+
+	// 4. Create HTTP server
+	app := server.NewServer(cfg.Port)
+
+	// 5. Initialize event consumer
+	eventConsumer, err := consumer.NewEventConsumer(
+		cfg.RedpandaBrokers,
+		cfg.ConsumerGroup,
+		cfg.RedpandaSASLUser,
+		cfg.RedpandaSASLPassword,
+		cfg.RedpandaSASLMechanism,
+		logger,
+	)
+	if err != nil {
+		logger.Error("failed to initialize consumer", "error", err)
+		return
+	}
+
+	var consumerWG sync.WaitGroup
+
+	// 6. Start event consumer in a goroutine
+	consumerWG.Go(func() {
+		if err := eventConsumer.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
+			logger.Error("consumer stopped with error", "error", err)
+			cancel()
+		}
+	})
+
+	// 7. Start HTTP server in a goroutine
+	go func() {
+		logger.Info("http server listening", "addr", cfg.Port)
+		if err := app.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Error("http server error", "error", err)
+			cancel()
+		}
+	}()
+
+	// 8. Mark service as ready
+	server.SetReady()
+	logger.Info("service is ready")
+
+	// 9. Wait for shutdown signal
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
+	defer signal.Stop(sigCh)
+	select {
+	case sig := <-sigCh:
+		logger.Info("received signal, shutting down", "signal", sig)
+	case <-ctx.Done():
+		logger.Info("context cancelled, shutting down")
+	}
+
+	// 10. Graceful shutdown
+	cancel()
+	gracefulShutdown(app, logger)
+	eventConsumer.Close()
+	consumerWG.Wait()
+
+	logger.Info("shutdown complete")
+}
+
+func gracefulShutdown(
+	app interface{ Shutdown(context.Context) error },
+	logger *slog.Logger,
+) {
+	server.SetNotReady()
+
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer shutdownCancel()
+
+	if err := app.Shutdown(shutdownCtx); err != nil {
+		logger.Error("http server shutdown error", "error", err)
+	}
+}

go.mod

@@ -1,22 +1,39 @@
-module github.com/Tanilytics/Processing
+module github.com/Tanilytics/processing
 
-go 1.25.0
+go 1.26.0
 
 require (
-	github.com/andybalholm/brotli v1.2.0 // indirect
-	github.com/gofiber/fiber/v3 v3.0.0 // indirect
-	github.com/gofiber/schema v1.6.0 // indirect
-	github.com/gofiber/utils/v2 v2.0.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
-	github.com/klauspost/compress v1.18.3 // indirect
-	github.com/mattn/go-colorable v0.1.14 // indirect
-	github.com/mattn/go-isatty v0.0.20 // indirect
-	github.com/philhofer/fwd v1.2.0 // indirect
-	github.com/tinylib/msgp v1.6.3 // indirect
-	github.com/valyala/bytebufferpool v1.0.0 // indirect
-	github.com/valyala/fasthttp v1.69.0 // indirect
-	golang.org/x/crypto v0.47.0 // indirect
-	golang.org/x/net v0.49.0 // indirect
-	golang.org/x/sys v0.40.0 // indirect
-	golang.org/x/text v0.33.0 // indirect
+	github.com/google/uuid v1.6.0
+	github.com/stretchr/testify v1.11.1
+	github.com/twmb/franz-go v1.20.7
+	go.opentelemetry.io/otel v1.42.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0
+	go.opentelemetry.io/otel/sdk v1.42.0
+)
+
+require (
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect
+	github.com/klauspost/compress v1.18.4 // indirect
+	github.com/pierrec/lz4/v4 v4.1.25 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect
+	go.opentelemetry.io/otel/metric v1.42.0 // indirect
+	go.opentelemetry.io/otel/trace v1.42.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
+	golang.org/x/text v0.34.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect
+	google.golang.org/grpc v1.79.2 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )