feat: add ch to ctrl

deployment/docker-compose.yaml

@@ -397,6 +397,8 @@ services:
       UNKEY_DEPOT_API_URL: "https://api.depot.dev"
       UNKEY_DEPOT_PROJECT_REGION: "us-east-1"
 
+      UNKEY_CLICKHOUSE_URL: "clickhouse://default:password@clickhouse:9000?secure=false&skip_verify=true"
+
   otel:
     networks:
       - default

go/apps/ctrl/config.go

@@ -142,6 +142,9 @@ type Config struct {
 	// BuildPlatform defines the target platform for builds (e.g., "linux/amd64", "linux/arm64")
 	BuildPlatform string
 	Depot         DepotConfig
+
+	// ClickhouseURL is the ClickHouse database connection string
+	ClickhouseURL string
 }
 
 type BuildPlatform struct {
@@ -198,6 +201,10 @@ func (c Config) Validate() error {
 		}
 	}
 
+	if err := assert.NotEmpty(c.ClickhouseURL, "ClickhouseURL is required"); err != nil {
+		return err
+	}
+
 	// Validate build platform format
 	_, platformErr := parseBuildPlatform(c.BuildPlatform)
 

go/apps/ctrl/run.go

@@ -26,6 +26,7 @@ import (
 	"github.com/unkeyed/unkey/go/gen/proto/ctrl/v1/ctrlv1connect"
 	hydrav1 "github.com/unkeyed/unkey/go/gen/proto/hydra/v1"
 	"github.com/unkeyed/unkey/go/gen/proto/krane/v1/kranev1connect"
+	"github.com/unkeyed/unkey/go/pkg/clickhouse"
 	"github.com/unkeyed/unkey/go/pkg/db"
 	"github.com/unkeyed/unkey/go/pkg/otel"
 	"github.com/unkeyed/unkey/go/pkg/otel/logging"
@@ -183,6 +184,17 @@ func Run(ctx context.Context, cfg Config) error {
 		return fmt.Errorf("unable to create build storage: %w", err)
 	}
 
+	var ch clickhouse.ClickHouse = clickhouse.NewNoop()
+	if cfg.ClickhouseURL != "" {
+		ch, err = clickhouse.New(clickhouse.Config{
+			URL:    cfg.ClickhouseURL,
+			Logger: logger,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to create clickhouse: %w", err)
+		}
+	}
+
 	var buildService ctrlv1connect.BuildServiceClient
 	switch cfg.BuildBackend {
 	case BuildBackendDocker:
@@ -202,6 +214,7 @@ func Run(ctx context.Context, cfg Config) error {
 			RegistryConfig: depot.RegistryConfig(cfg.GetRegistryConfig()),
 			BuildPlatform:  depot.BuildPlatform(cfg.GetBuildPlatform()),
 			DepotConfig:    depot.DepotConfig(cfg.GetDepotConfig()),
+			Clickhouse:     ch,
 			Logger:         logger,
 			Storage:        buildStorage,
 		})

go/apps/ctrl/services/build/backend/depot/service.go

@@ -4,6 +4,7 @@ package depot
 import (
 	"github.com/unkeyed/unkey/go/apps/ctrl/services/build/storage"
 	"github.com/unkeyed/unkey/go/gen/proto/ctrl/v1/ctrlv1connect"
+	"github.com/unkeyed/unkey/go/pkg/clickhouse"
 	"github.com/unkeyed/unkey/go/pkg/db"
 	"github.com/unkeyed/unkey/go/pkg/otel/logging"
 )
@@ -32,6 +33,7 @@ type Depot struct {
 	depotConfig    DepotConfig
 	registryConfig RegistryConfig
 	buildPlatform  BuildPlatform
+	clickhouse     clickhouse.ClickHouse
 	logger         logging.Logger
 }
 
@@ -40,6 +42,7 @@ type Config struct {
 	DB             db.Database
 	Storage        *storage.S3
 	DepotConfig    DepotConfig
+	Clickhouse     clickhouse.ClickHouse // Clickhouse for telemetry
 	RegistryConfig RegistryConfig
 	BuildPlatform  BuildPlatform
 	Logger         logging.Logger
@@ -52,6 +55,7 @@ func New(cfg Config) *Depot {
 		db:                               cfg.DB,
 		storage:                          cfg.Storage,
 		depotConfig:                      cfg.DepotConfig,
+		clickhouse:                       cfg.Clickhouse,
 		registryConfig:                   cfg.RegistryConfig,
 		buildPlatform:                    cfg.BuildPlatform,
 		logger:                           cfg.Logger,

go/cmd/ctrl/main.go

@@ -117,6 +117,8 @@ var Cmd = &cli.Command{
 			cli.Default(9080), cli.EnvVar("UNKEY_RESTATE_HTTP_PORT")),
 		cli.String("restate-register-as", "URL of this service for self-registration with Restate. Example: http://ctrl:9080",
 			cli.EnvVar("UNKEY_RESTATE_REGISTER_AS")),
+		cli.String("clickhouse-url", "ClickHouse connection string for analytics. Recommended for production. Example: clickhouse://user:pass@host:9000/unkey",
+			cli.EnvVar("UNKEY_CLICKHOUSE_URL")),
 	},
 	Action: action,
 }
@@ -213,6 +215,9 @@ func action(ctx context.Context, cmd *cli.Command) error {
 			RegisterAs: cmd.String("restate-register-as"),
 		},
 
+		// Clickhouse Configuration
+		ClickhouseURL: cmd.String("clickhouse-url"),
+
 		// Common
 		Clock: clock.New(),
 	}

go/pkg/clickhouse/client.go

@@ -379,6 +379,14 @@ func (c *clickhouse) BufferRatelimit(req schema.RatelimitRequestV1) {
 	c.ratelimits.Buffer(req)
 }
 
+func (c *clickhouse) BufferBuildStep(req schema.BuildStepV1) {
+	c.buildSteps.Buffer(req)
+}
+
+func (c *clickhouse) BufferBuildStepLog(req schema.BuildStepLogV1) {
+	c.buildStepLogs.Buffer(req)
+}
+
 func (c *clickhouse) Conn() ch.Conn {
 	return c.conn
 }

go/pkg/clickhouse/interface.go

@@ -33,6 +33,14 @@ type Bufferer interface {
 	// BufferRatelimit adds a ratelimit event to the buffer.
 	// These represent API ratelimit operations with their outcome.
 	BufferRatelimit(schema.RatelimitRequestV1)
+
+	// BufferRatelimit adds a ratelimit event to the buffer.
+	// These represent API ratelimit operations with their outcome.
+	BufferBuildStep(schema.BuildStepV1)
+
+	// BufferRatelimit adds a ratelimit event to the buffer.
+	// These represent API ratelimit operations with their outcome.
+	BufferBuildStepLog(schema.BuildStepLogV1)
 }
 
 type Querier interface {

go/pkg/clickhouse/noop.go

@@ -12,8 +12,10 @@ import (
 // such as in development environments or when running integration tests.
 type noop struct{}
 
-var _ Bufferer = (*noop)(nil)
-var _ Bufferer = (*noop)(nil)
+var (
+	_ Bufferer = (*noop)(nil)
+	_ Bufferer = (*noop)(nil)
+)
 
 // BufferRequest implements the Bufferer interface but discards the event.
 func (n *noop) BufferRequest(schema.ApiRequestV1) {
@@ -40,6 +42,16 @@ func (n *noop) BufferRatelimit(req schema.RatelimitRequestV1) {
 	// Intentionally empty - discards the event
 }
 
+// BufferBuildStep implements the Bufferer interface but discards the event.
+func (n *noop) BufferBuildStep(req schema.BuildStepV1) {
+	// Intentionally empty - discards the event
+}
+
+// BufferBuildStepLog implements the Bufferer interface but discards the event.
+func (n *noop) BufferBuildStepLog(req schema.BuildStepLogV1) {
+	// Intentionally empty - discards the event
+}
+
 // GetBillableVerifications implements the Bufferer interface but always returns 0.
 func (n *noop) GetBillableVerifications(ctx context.Context, workspaceID string, year, month int) (int64, error) {
 	return 0, nil