feat: nats event publisher + user events (#1629)

Signed-off-by: Miguel Martinez <[email protected]>

Author: migmartri

app/controlplane/cmd/main.go

@@ -17,6 +17,7 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"time"
 
@@ -25,6 +26,7 @@ import (
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/ca/ejbca"
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/ca/fileca"
 	"github.com/getsentry/sentry-go"
+	"github.com/nats-io/nats.go"
 	flag "github.com/spf13/pflag"
 
 	conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1"
@@ -145,6 +147,7 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
+
 	app, cleanup, err := wireApp(&bc, credsWriter, logger, availablePlugins, ca)
 	if err != nil {
 		panic(err)
@@ -178,6 +181,21 @@ type app struct {
 	tokenAuthSyncer  *biz.APITokenSyncerUseCase
 }
 
+// Connection to nats is optional, if not configured, pubsub will be disabled
+func newNatsConnection(c *conf.Bootstrap_NatsServer) (*nats.Conn, error) {
+	uri := c.GetUri()
+	if uri == "" {
+		return nil, nil
+	}
+
+	nc, err := nats.Connect(uri)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to nats: %w", err)
+	}
+
+	return nc, nil
+}
+
 func filterSensitiveArgs(_ log.Level, keyvals ...interface{}) bool {
 	for i := 0; i < len(keyvals); i++ {
 		if keyvals[i] == "operation" {

app/controlplane/cmd/wire.go

@@ -1,5 +1,5 @@
 //
-// Copyright 2023 The Chainloop Authors.
+// Copyright 2024 The Chainloop Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import (
 	"github.com/chainloop-dev/chainloop/app/controlplane/internal/dispatcher"
 	"github.com/chainloop-dev/chainloop/app/controlplane/internal/server"
 	"github.com/chainloop-dev/chainloop/app/controlplane/internal/service"
+	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor"
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/authz"
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/biz"
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/ca"
@@ -49,7 +50,7 @@ func wireApp(*conf.Bootstrap, credentials.ReaderWriter, log.Logger, sdk.Availabl
 			wire.Bind(new(biz.CASClient), new(*biz.CASClientUseCase)),
 			serviceOpts,
 			wire.Value([]biz.CASClientOpts{}),
-			wire.FieldsOf(new(*conf.Bootstrap), "Server", "Auth", "Data", "CasServer", "ReferrerSharedIndex", "Onboarding", "PrometheusIntegration", "PolicyProviders"),
+			wire.FieldsOf(new(*conf.Bootstrap), "Server", "Auth", "Data", "CasServer", "ReferrerSharedIndex", "Onboarding", "PrometheusIntegration", "PolicyProviders", "NatsServer"),
 			wire.FieldsOf(new(*conf.Data), "Database"),
 			dispatcher.New,
 			authz.NewDatabaseEnforcer,
@@ -58,6 +59,8 @@ func wireApp(*conf.Bootstrap, credentials.ReaderWriter, log.Logger, sdk.Availabl
 			newProtoValidator,
 			newDataConf,
 			newPolicyProviderConfig,
+			newNatsConnection,
+			auditor.NewAuditLogPublisher,
 		),
 	)
 }

app/controlplane/cmd/wire_gen.go

@@ -16,6 +16,9 @@ server:
     #   certificate: "../../devel/devkeys/selfsigned/controlplane.crt"
     #   private_key: "../../devel/devkeys/selfsigned/controlplane.key"
 
+nats_server:
+  uri: nats://0.0.0.0:4222
+
 certificate_authority:
   file_ca:
     cert_path: ${FILE_CA_CERT_PATH:../../devel/devkeys/ca.pub}

app/controlplane/configs/config.devel.yaml

@@ -30,6 +30,10 @@ cas_server:
   # Where to redirect the user to download artifacts from the CAS
   download_url: http://0.0.0.0:8001/download
 
+# nats endpoint where to send events
+nats_server:
+  uri: nats://0.0.0.0:4222
+
 # Where to store credentials such as OCI registries or third party integrations secrets
 credentials_service:
   vault: