feat(reconciliation): add policies and more endpoints (#963)

Earthfile

@@ -32,7 +32,7 @@ build-final-spec:
         COPY components/$c/openapi.yaml $c/openapi.yaml
     END
     WORKDIR /src/ee
-    FOR c IN auth webhooks search wallets orchestration
+    FOR c IN auth webhooks search wallets orchestration reconciliation
         COPY ee/$c/openapi.yaml $c/openapi.yaml
     END
     WORKDIR /src/openapi

components/operator/.earthly/configuration/templates/configuration.yaml

@@ -57,6 +57,11 @@ spec:
         disableSSLMode: true
     reconciliation:
       debug: true
+      postgres:
+        credentialsFromSecret: postgres
+        host: postgres-postgresql.formance.svc.cluster.local
+        port: 5432
+        disableSSLMode: true
     payments:
       postgres:
         credentialsFromSecret: postgres
@@ -71,7 +76,6 @@ spec:
         port: 5432
         disableSSLMode: true
     search:
-
       elasticSearch:
         host: elasticsearch-master.formance.svc.cluster.local
         port: 9200

components/operator/apis/stack/v1beta3/reconciliation_types.go

@@ -2,6 +2,7 @@ package v1beta3
 
 type ReconciliationSpec struct {
 	CommonServiceProperties `json:",inline"`
+	Postgres                PostgresConfig `json:"postgres"`
 
 	// +optional
 	ResourceProperties *ResourceProperties     `json:"resourceProperties,omitempty"`

components/operator/internal/modules/module.go

@@ -26,6 +26,8 @@ type Cron struct {
 }
 
 type DatabaseMigration struct {
+	// optional, will be the commit hash if empty in the kube object
+	Name          string
 	Shutdown      bool
 	Command       []string
 	AdditionalEnv func(config ReconciliationConfig) []EnvVar
@@ -357,7 +359,11 @@ func (r *moduleReconciler) runPreUpgradeMigration(ctx context.Context, module Mo
 
 func (r *moduleReconciler) runDatabaseMigration(ctx context.Context, version string, migration DatabaseMigration, postgresConfig v1beta3.PostgresConfig) (bool, error) {
 	logger := log.FromContext(ctx)
-	return r.RunJob(ctx, fmt.Sprintf("%s-%s-db-migration", r.module.Name(), version),
+	name := fmt.Sprintf("%s-%s-db-migration", r.module.Name(), version)
+	if migration.Name != "" {
+		name = fmt.Sprintf("%s-%s-db-migration", r.module.Name(), migration.Name)
+	}
+	return r.RunJob(ctx, name,
 		func() error {
 			if migration.Shutdown {
 				logger.Info("Stop module reconciliation as required by upgrade", "module", r.module.Name())

components/operator/internal/modules/reconciliation/handler.go

@@ -1,6 +1,7 @@
 package reconciliation
 
 import (
+	"github.com/formancehq/operator/apis/stack/v1beta3"
 	stackv1beta3 "github.com/formancehq/operator/apis/stack/v1beta3"
 	"github.com/formancehq/operator/internal/modules"
 )
@@ -15,13 +16,25 @@ func (o module) IsEE() bool {
 	return true
 }
 
+func (p module) Postgres(ctx modules.ReconciliationConfig) v1beta3.PostgresConfig {
+	return ctx.Configuration.Spec.Services.Reconciliation.Postgres
+}
+
 func (o module) Versions() map[string]modules.Version {
 	return map[string]modules.Version{
 		"v0.0.0": {
+			DatabaseMigration: &modules.DatabaseMigration{
+				Name:     "v0.0.0",
+				Shutdown: false,
+				Command:  []string{"migrate", "up"},
+				AdditionalEnv: func(config modules.ReconciliationConfig) []modules.EnvVar {
+					return []modules.EnvVar{}
+				},
+			},
 			Services: func(ctx modules.ReconciliationConfig) modules.Services {
 				return modules.Services{
 					{
-						InjectPostgresVariables: false,
+						InjectPostgresVariables: true,
 						AuthConfiguration: func(config modules.ReconciliationConfig) stackv1beta3.ClientConfiguration {
 							return stackv1beta3.NewClientConfiguration()
 						},
@@ -48,6 +61,7 @@ func (o module) Versions() map[string]modules.Version {
 
 func reconciliationEnvVars(resolveContext modules.ContainerResolutionConfiguration) modules.ContainerEnv {
 	env := modules.NewEnv().Append(
+		modules.Env("POSTGRES_DATABASE_NAME", "$(POSTGRES_DATABASE)"),
 		modules.Env("STACK_CLIENT_ID", resolveContext.Stack.Status.StaticAuthClients["reconciliation"].ID),
 		modules.Env("STACK_CLIENT_SECRET", resolveContext.Stack.Status.StaticAuthClients["reconciliation"].Secrets[0]),
 	)
@@ -58,6 +72,7 @@ func reconciliationEnvVars(resolveContext modules.ContainerResolutionConfigurati
 var Module = &module{}
 
 var _ modules.Module = Module
+var _ modules.PostgresAwareModule = Module
 
 func init() {
 	modules.Register(Module)

ee/reconciliation/.gitignore

@@ -7,4 +7,4 @@ vendor
 coverage.out
 dist/
 .env
-payments
+reconciliation

ee/reconciliation/cmd/migrate.go

@@ -1,6 +1,18 @@
 package cmd
 
-import "github.com/spf13/cobra"
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	storage "github.com/formancehq/reconciliation/internal/storage/migrations"
+	"github.com/formancehq/stack/libs/go-libs/service"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/dialect/pgdialect"
+	"github.com/uptrace/bun/extra/bundebug"
+)
 
 func newMigrate() *cobra.Command {
 	migrate := &cobra.Command{
@@ -13,5 +25,32 @@ func newMigrate() *cobra.Command {
 }
 
 func runMigrate(cmd *cobra.Command, args []string) error {
-	return nil
+	postgresURI := viper.GetString(postgresURIFlag)
+	if postgresURI == "" {
+		postgresURI = cmd.Flag(postgresURIFlag).Value.String()
+	}
+
+	if postgresURI == "" {
+		return fmt.Errorf("postgres uri is not set")
+	}
+
+	// TODO: Maybe use pgx everywhere instead of pq
+	db, err := sql.Open("postgres", postgresURI)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = db.Close()
+	}()
+
+	bunDB := bun.NewDB(db, pgdialect.New())
+	if viper.GetBool(service.DebugFlag) {
+		bunDB.AddQueryHook(bundebug.NewQueryHook(bundebug.WithWriter(cmd.OutOrStdout())))
+	}
+
+	return Migrate(cmd.Context(), bunDB)
+}
+
+func Migrate(ctx context.Context, db *bun.DB) error {
+	return storage.Migrate(ctx, db)
 }

ee/reconciliation/cmd/root.go

@@ -20,6 +20,7 @@ const (
 	stackClientIDFlag     = "stack-client-id"
 	stackClientSecretFlag = "stack-client-secret"
 	listenFlag            = "listen"
+	postgresURIFlag       = "postgres-uri"
 )
 
 func NewRootCommand() *cobra.Command {

ee/reconciliation/cmd/serve.go

@@ -2,11 +2,14 @@ package cmd
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 
 	sdk "github.com/formancehq/formance-sdk-go"
 	"github.com/formancehq/reconciliation/internal/api"
+	"github.com/formancehq/reconciliation/internal/storage"
 	sharedapi "github.com/formancehq/stack/libs/go-libs/api"
 	"github.com/formancehq/stack/libs/go-libs/otlp"
 	"github.com/formancehq/stack/libs/go-libs/otlp/otlpmetrics"
@@ -55,7 +58,13 @@ func newServeCommand(version string) *cobra.Command {
 
 func runServer(version string) func(cmd *cobra.Command, args []string) error {
 	return func(cmd *cobra.Command, args []string) error {
+		databaseOptions, err := prepareDatabaseOptions(cmd.OutOrStdout())
+		if err != nil {
+			return err
+		}
+
 		options := make([]fx.Option, 0)
+		options = append(options, databaseOptions)
 
 		options = append(options,
 			otlptraces.CLITracesModule(viper.GetViper()),
@@ -72,3 +81,12 @@ func runServer(version string) func(cmd *cobra.Command, args []string) error {
 		return service.New(cmd.OutOrStdout(), options...).Run(cmd.Context())
 	}
 }
+
+func prepareDatabaseOptions(output io.Writer) (fx.Option, error) {
+	postgresURI := viper.GetString(postgresURIFlag)
+	if postgresURI == "" {
+		return nil, errors.New("missing postgres uri")
+	}
+
+	return storage.Module(postgresURI, viper.GetBool(service.DebugFlag), output), nil
+}

ee/reconciliation/go.mod

@@ -8,10 +8,17 @@ require (
 	github.com/go-chi/chi/v5 v5.0.10
 	github.com/golang/mock v1.4.4
 	github.com/google/uuid v1.3.0
+	github.com/jackc/pgconn v1.14.1
+	github.com/jackc/pgx/v5 v5.3.0
+	github.com/pkg/errors v0.9.1
 	github.com/riandyrn/otelchi v0.5.1
 	github.com/spf13/cobra v1.8.0
 	github.com/spf13/viper v1.17.0
 	github.com/stretchr/testify v1.8.4
+	github.com/uptrace/bun v1.1.16
+	github.com/uptrace/bun/dialect/pgdialect v1.1.16
+	github.com/uptrace/bun/extra/bundebug v1.1.16
+	github.com/uptrace/bun/extra/bunotel v1.1.16
 	go.uber.org/fx v1.20.1
 	golang.org/x/oauth2 v0.12.0
 	golang.org/x/sync v0.3.0
@@ -22,6 +29,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/ericlagergren/decimal v0.0.0-20221120152707-495c53812d05 // indirect
+	github.com/fatih/color v1.15.0 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
@@ -31,13 +39,21 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.2 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/lib/pq v1.10.7 // indirect
 	github.com/lithammer/shortuuid/v3 v3.0.7 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/sagikazarmark/locafero v0.3.0 // indirect
@@ -52,8 +68,12 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.11 // indirect
 	github.com/tklauser/numcpus v0.6.0 // indirect
+	github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc // indirect
 	github.com/uptrace/opentelemetry-go-extra/otellogrus v0.1.21 // indirect
+	github.com/uptrace/opentelemetry-go-extra/otelsql v0.2.2 // indirect
 	github.com/uptrace/opentelemetry-go-extra/otelutil v0.1.21 // indirect
+	github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
+	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.opentelemetry.io/contrib v1.0.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/host v0.42.0 // indirect
@@ -80,6 +100,7 @@ require (
 	go.uber.org/dig v1.17.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
+	golang.org/x/crypto v0.14.0 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
 	golang.org/x/net v0.15.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect