stack/components/operator/internal/modules/gateway/handler.go

package gateway

import (
	"bytes"
	"fmt"
	"sort"
	"strings"
	"text/template"

	"github.com/formancehq/operator/internal/modules/auth"
	"github.com/formancehq/operator/internal/modules/control"
	"github.com/formancehq/operator/internal/modules/ledger"
	"github.com/formancehq/operator/internal/modules/orchestration"
	"github.com/formancehq/operator/internal/modules/payments"
	"github.com/formancehq/operator/internal/modules/reconciliation"
	"github.com/formancehq/operator/internal/modules/search"
	"github.com/formancehq/operator/internal/modules/wallets"
	"github.com/formancehq/operator/internal/modules/webhooks"
	"golang.org/x/mod/semver"

	"github.com/formancehq/operator/internal/modules"
)

const (
	gatewayPort = 8000
)

type module struct{}

func (g module) Name() string {
	return "gateway"
}

func (g module) DependsOn() []modules.Module {
	return []modules.Module{
		auth.Module,
		control.Module,
		ledger.Module,
		orchestration.Module,
		reconciliation.Module,
		payments.Module,
		search.Module,
		wallets.Module,
		webhooks.Module,
	}
}

func (g module) Versions() map[string]modules.Version {
	return map[string]modules.Version{
		"v0.0.0": {
			Services: func(ctx modules.ReconciliationConfig) modules.Services {
				service := &modules.Service{
					Port: gatewayPort,
					ExposeHTTP: &modules.ExposeHTTP{
						Path: "/",
					},
					Liveness:         modules.LivenessDisable,
					LivenessEndpoint: ctx.Configuration.Spec.Services.Gateway.LivenessEndpoint,
					Topics:           &modules.Topics{Name: "audit"},
					Annotations:      ctx.Configuration.Spec.Services.Gateway.Annotations.Service,
					Configs: func(resolveContext modules.ServiceInstallConfiguration) modules.Configs {
						return modules.Configs{
							"config": modules.Config{
								Data: map[string]string{
									"Caddyfile": createCaddyfile(resolveContext),
								},
								Mount: true,
							},
						}
					},
					Container: func(resolveContext modules.ContainerResolutionConfiguration) modules.Container {
						return modules.Container{
							Command: []string{"/usr/bin/caddy"},
							Args: []string{
								"run",
								"--config", resolveContext.GetConfig("config").GetMountPath() + "/Caddyfile",
								"--adapter", "caddyfile",
							},
							Image: modules.GetImage("gateway", resolveContext.Versions.Spec.Gateway),
							Env: modules.NewEnv().
								Append(modules.BrokerEnvVarsWithPrefix(resolveContext.Configuration.Spec.Broker, "gateway")...),
							Resources: modules.GetResourcesWithDefault(
								resolveContext.Configuration.Spec.Services.Gateway.ResourceProperties,
								modules.ResourceSizeSmall(),
							),
						}
					},
				}

				return modules.Services{service}
			},
		},
	}
}

var Module = &module{}

var _ modules.Module = Module
var _ modules.DependsOnAwareModule = Module

func init() {
	modules.Register(Module)
}

func caddyFileTemplate(context modules.ServiceInstallConfiguration) *template.Template {
	gatewayVersion := context.Versions.Spec.Gateway
	if !semver.IsValid(gatewayVersion) {
		return template.Must(template.New("caddyfile").Funcs(map[string]any{
			"join": strings.Join,
		}).Parse(caddyfile))
	}

	switch semver.Compare("v0.1.8", gatewayVersion) {
	case -1, 0:
		// gatewayVersion >= v0.1.8
		return template.Must(template.New("caddyfile").Funcs(map[string]any{
			"join": strings.Join,
		}).Parse(caddyfile))
	default:
		// gatewayVersion < v0.1.8
		return template.Must(template.New("caddyfile").Funcs(map[string]any{
			"join": strings.Join,
		}).Parse(deprecatedCaddyfile))
	}
}

// When enabled, the audit plugin will be enabled for the gateway.
// The audit plugin is available since gateway v0.2.0.
// If the gateway version is not available, the audit plugin will be disabled.
// Also if not enabled the nats stream will not be created.
func EnableAuditPlugin(ctx modules.ReconciliationConfig) bool {
	gatewayVersion := ctx.Versions.Spec.Gateway
	enable := ctx.Configuration.Spec.Services.Gateway.EnableAuditPlugin
	if enable == nil {
		return false
	}

	if !semver.IsValid(gatewayVersion) {
		return *enable
	}

	return *enable && semver.Compare("v0.2.0", gatewayVersion) <= 0
}

func createCaddyfile(context modules.ServiceInstallConfiguration) string {
	buf := bytes.NewBufferString("")

	type service struct {
		modules.RegisteredService
		Name        string
		Port        int32
		Hostname    string
		HealthPath  string
		Methods     []string
		RoutingPath string
		Module      string
	}

	type versionEndpoint struct {
		VersionEndpoint string
		HealthEndpoint  string
	}

	type versions struct {
		ModuleName string
		Endpoints  []versionEndpoint
	}

	servicesMap := make(map[string]service, 0)
	keys := make([]string, 0)
	for _, registeredModule := range context.RegisteredModules {
		if registeredModule.Module.Name() == "gateway" {
			continue
		}
		for _, s := range registeredModule.Services {
			if s.ExposeHTTP == nil {
				continue
			}
			usedPort := s.Port
			if usedPort == 0 {
				continue
			}
			serviceName := registeredModule.Module.Name()
			if s.Name != "" {
				serviceName += "-" + s.Name
			}
			serviceRoutingPath := registeredModule.Module.Name()
			if s.ExposeHTTP.Name != "" {
				serviceRoutingPath = serviceRoutingPath + "-" + s.ExposeHTTP.Name
			}

			healthPath := "_healthcheck"
			if s.Liveness == modules.LivenessLegacy {
				healthPath = "_health"
			}
			hostname := serviceName
			if context.Configuration.Spec.LightMode {
				hostname = "127.0.0.1"
			}

			servicesMap[serviceName] = service{
				Name:              serviceName,
				RegisteredService: s,
				Port:              usedPort,
				Hostname:          hostname,
				HealthPath:        healthPath,
				Methods:           s.ExposeHTTP.Methods,
				RoutingPath:       serviceRoutingPath,
				Module:            registeredModule.Module.Name(),
			}
			keys = append(keys, serviceName)
		}
	}

	sort.Strings(keys)
	services := make([]service, 0)
	for _, key := range keys {
		services = append(services, servicesMap[key])
	}

	fallback := ""
	redirect := false
	if context.Configuration.Spec.Services.Gateway.Fallback != nil && *context.Configuration.Spec.Services.Gateway.Fallback != "" {
		fallback = *context.Configuration.Spec.Services.Gateway.Fallback
		redirect = true
	} else {
		if !context.IsDisabled("control", false) {
			fallback = fmt.Sprintf("control:%d", servicesMap["control"].Port)
		}
	}
	enableScopes := false
	if context.ReconciliationConfig.Configuration.Spec.Services.Gateway.EnableScopes != nil {
		enableScopes = *context.ReconciliationConfig.Configuration.Spec.Services.Gateway.EnableScopes
	}

	versionsMap := make(map[string]*versions, 0)
	for _, service := range services {
		if !service.HasVersionEndpoint {
			continue
		}

		_, ok := versionsMap[service.Module]
		if !ok {
			versionsMap[service.Module] = &versions{
				ModuleName: service.Module,
				Endpoints:  make([]versionEndpoint, 0),
			}
		}

		versionsMap[service.Module].Endpoints = append(versionsMap[service.Module].Endpoints, versionEndpoint{
			VersionEndpoint: fmt.Sprintf("http://%s:%d/_info", service.Hostname, service.Port),
			HealthEndpoint:  fmt.Sprintf("http://%s:%d/%s", service.Hostname, service.Port, service.HealthPath),
		})
	}

	vs := make([]*versions, 0)
	for _, v := range versionsMap {
		vs = append(vs, v)
	}

	sort.Slice(vs, func(i, j int) bool {
		return vs[i].ModuleName < vs[j].ModuleName
	})

	issuer := fmt.Sprintf("%s/api/auth", context.Stack.URL())
	if semver.IsValid(context.Versions.Spec.Gateway) {
		switch semver.Compare(context.Versions.Spec.Gateway, "v2.0.0-alpha") {
		case -1:
			issuer = fmt.Sprintf("%s/api/auth", context.Stack.PublicURL())
		}
	}

	if err := caddyFileTemplate(context).Execute(buf, map[string]any{
		"Region":   context.Platform.Region,
		"Env":      context.Platform.Environment,
		"Issuer":   issuer,
		"Services": services,
		"Debug":    context.Stack.Spec.Debug,
		"Broker": func() string {
			if context.Configuration.Spec.Broker.Kafka != nil {
				return "kafka"
			}
			if context.Configuration.Spec.Broker.Nats != nil {
				return "nats"
			}
			return ""
		}(),
		"Port":         gatewayPort,
		"Fallback":     fallback,
		"Redirect":     redirect,
		"EnableAudit":  EnableAuditPlugin(context.ReconciliationConfig),
		"EnableScopes": enableScopes,
		"Versions":     vs,
	}); err != nil {
		panic(err)
	}
	return buf.String()
}

// TODO: remove auth part when everything is migrated to the new auth middleware
// per service

const caddyfile = `(cors) {
	header {
		Access-Control-Allow-Methods "GET,OPTIONS,PUT,POST,DELETE,HEAD,PATCH"
		Access-Control-Allow-Headers content-type
		Access-Control-Max-Age 100
		Access-Control-Allow-Origin *
	}
}

(auth) {
	auth {
		issuer {{ .Issuer }}

		read_key_set_max_retries 10

		{{- if .EnableScopes }}
		check_scopes yes
		service {args[0]}
		{{- end }}
	}
}
{{- if .EnableAudit }}
(audit) {
	audit {
		# Kafka publisher
		{{- if (eq .Broker "kafka") }}
		publisher_kafka_broker {$PUBLISHER_KAFKA_BROKER:redpanda:29092}
		publisher_kafka_enabled {$PUBLISHER_KAFKA_ENABLED:false}
		publisher_kafka_tls_enabled {$PUBLISHER_KAFKA_TLS_ENABLED:false}
		publisher_kafka_sasl_enabled {$PUBLISHER_KAFKA_SASL_ENABLED:false}
		publisher_kafka_sasl_username {$PUBLISHER_KAFKA_SASL_USERNAME}
		publisher_kafka_sasl_password {$PUBLISHER_KAFKA_SASL_PASSWORD}
		publisher_kafka_sasl_mechanism {$PUBLISHER_KAFKA_SASL_MECHANISM}
		publisher_kafka_sasl_scram_sha_size {$PUBLISHER_KAFKA_SASL_SCRAM_SHA_SIZE}
		{{- end }}
		{{- if (eq .Broker "nats") }}
		# Nats publisher
		publisher_nats_enabled {$PUBLISHER_NATS_ENABLED:true}
		publisher_nats_url {$PUBLISHER_NATS_URL:nats://nats:4222}
		publisher_nats_client_id {$PUBLISHER_NATS_CLIENT_ID:gateway}
		publisher_nats_max_reconnects {$PUBLISHER_NATS_MAX_RECONNECTS:30}
		publisher_nats_max_reconnects_wait {$PUBLISHER_NATS_MAX_RECONNECT_WAIT:2s}
		{{- end }}
	}
}
{{- end }}

{
	{{ if .Debug }}debug{{ end }}
	# Many directives manipulate the HTTP handler chain and the order in which
	# those directives are evaluated matters. So the jwtauth directive must be
	# ordered.
	# c.f. https://caddyserver.com/docs/caddyfile/directives#directive-order
	order auth before basicauth
	order versions after metrics
	{{- if .EnableAudit }}
	order audit after encode
	{{- end }}
}

:{{ .Port }} {
	tracing {
		span gateway
	}
	log {
		output stdout
		{{- if .Debug }}
		level  DEBUG
		{{- end }}
	}

	{{- if .EnableAudit }}
	import audit
	{{- end }}

	{{- range $i, $service := .Services }}
		{{- if not (eq $service.Name "control") }}
			{{- range $i, $path := $service.Paths }}
			@{{ $path.Name }}matcher {
				path /api/{{ $service.RoutingPath }}{{ $path.Path }}*
				{{- if gt ($path.Methods | len) 0 }}
				method {{ join $path.Methods " " }}
				{{- end }}
			}
			handle @{{ $path.Name }}matcher {
				uri strip_prefix /api/{{ $service.RoutingPath }}
				reverse_proxy {{ $service.Hostname }}:{{ $service.Port }}
				import cors
				{{- if and (not $service.Secured) (not $path.Secured) }}
				import auth {{ $service.Module }}
				{{- end }}
			}
			{{- end }}

			@{{ $service.Name }}matcher {
				path /api/{{ $service.RoutingPath }}*
				{{- if gt ($service.Methods | len) 0 }}
				method {{ join $service.Methods " " }}
				{{- end }}
			}
			handle @{{ $service.Name }}matcher {
				uri strip_prefix /api/{{ $service.RoutingPath }}
				reverse_proxy {{ $service.Hostname }}:{{ $service.Port }}

				import cors
				{{- if not $service.Secured }}
				import auth {{ $service.Module }}
				{{- end }}
			}
		{{- end }}
	{{- end }}

	handle /versions {
		versions {
			region "{{ .Region }}"
			env "{{ .Env }}"
			endpoints {
				{{- range $i, $version := .Versions }}
					{{ $version.ModuleName }} {
						{{- range $i, $endpoint := $version.Endpoints }}
						{{ $endpoint.VersionEndpoint }} {{ $endpoint.HealthEndpoint }}
						{{- end }}
					}
				{{- end }}
			}
		}
	}

	# Respond 502 if service does not exists
	handle /api/* {
		respond "Bad Gateway" 502
	}

	# handle all other requests
	{{- if not (eq .Fallback "") }}
	handle {
		{{- if .Redirect }}
		redir {{ .Fallback }}
		{{- else }}
		reverse_proxy {{ .Fallback }}
		{{- end }}
		import cors
	}
	{{ end }}
}`

const deprecatedCaddyfile = `(cors) {
	header {
		Access-Control-Allow-Methods "GET,OPTIONS,PUT,POST,DELETE,HEAD,PATCH"
		Access-Control-Allow-Headers content-type
		Access-Control-Max-Age 100
		Access-Control-Allow-Origin *
	}
}

(auth) {
	auth {
		issuer {{ .Issuer }}

		read_key_set_max_retries 10

		{{- if .EnableScopes }}
		check_scopes yes
		service {args[0]}
		{{- end }}
	}
}
{{- if .EnableAudit }}
(audit) {
	audit {
		# Kafka publisher
		{{- if (eq .Broker "kafka") }}
		publisher_kafka_broker {$PUBLISHER_KAFKA_BROKER:redpanda:29092}
		publisher_kafka_enabled {$PUBLISHER_KAFKA_ENABLED:false}
		publisher_kafka_tls_enabled {$PUBLISHER_KAFKA_TLS_ENABLED:false}
		publisher_kafka_sasl_enabled {$PUBLISHER_KAFKA_SASL_ENABLED:false}
		publisher_kafka_sasl_username {$PUBLISHER_KAFKA_SASL_USERNAME}
		publisher_kafka_sasl_password {$PUBLISHER_KAFKA_SASL_PASSWORD}
		publisher_kafka_sasl_mechanism {$PUBLISHER_KAFKA_SASL_MECHANISM}
		publisher_kafka_sasl_scram_sha_size {$PUBLISHER_KAFKA_SASL_SCRAM_SHA_SIZE}
		{{- end }}
		{{- if (eq .Broker "nats") }}
		# Nats publisher
		publisher_nats_enabled {$PUBLISHER_NATS_ENABLED:true}
		publisher_nats_url {$PUBLISHER_NATS_URL:nats://nats:4222}
		publisher_nats_client_id {$PUBLISHER_NATS_CLIENT_ID:gateway}
		publisher_nats_max_reconnects {$PUBLISHER_NATS_MAX_RECONNECTS:30}
		publisher_nats_max_reconnects_wait {$PUBLISHER_NATS_MAX_RECONNECT_WAIT:2s}
		{{- end }}
	}
}
{{- end }}

{
	{{ if .Debug }}debug{{ end }}
	# Many directives manipulate the HTTP handler chain and the order in which
	# those directives are evaluated matters. So the jwtauth directive must be
	# ordered.
	# c.f. https://caddyserver.com/docs/caddyfile/directives#directive-order
	order auth before basicauth
	order versions after metrics
	{{- if .EnableAudit }}
	order audit after encode
	{{- end }}
}

:{{ .Port }} {
	tracing {
		span gateway
	}
	log {
		output stdout
		{{- if .Debug }}
		level  DEBUG
		{{- end }}
	}

	{{- if .EnableAudit }}
	import audit
	{{- end }}

	{{- range $i, $service := .Services }}
		{{- if not (eq $service.Name "control") }}
			{{- range $i, $path := $service.Paths }}
			@{{ $path.Name }}matcher {
				path /api/{{ $service.RoutingPath }}{{ $path.Path }}*
				{{- if gt ($path.Methods | len) 0 }}
				method {{ join $path.Methods " " }}
				{{- end }}
			}
			handle @{{ $path.Name }}matcher {
				uri strip_prefix /api/{{ $service.RoutingPath }}
				reverse_proxy {{ $service.Hostname }}:{{ $service.Port }}
				import cors
				{{- if and (not $service.Secured) (not $path.Secured) }}
				import auth {{ $service.Module }}
				{{- end }}
			}
			{{- end }}

			@{{ $service.Name }}matcher {
				path /api/{{ $service.RoutingPath }}*
				{{- if gt ($service.Methods | len) 0 }}
				method {{ join $service.Methods " " }}
				{{- end }}
			}
			handle @{{ $service.Name }}matcher {
				uri strip_prefix /api/{{ $service.RoutingPath }}
				reverse_proxy {{ $service.Hostname }}:{{ $service.Port }}

				import cors
				{{- if not $service.Secured }}
				import auth {{ $service.Module }}
				{{- end }}
			}
		{{- end }}
	{{- end }}

	handle /versions {
		versions {
			region "{{ .Region }}"
			env "{{ .Env }}"
			endpoints {
				{{- range $i, $service := .Services }}
					{{- if $service.HasVersionEndpoint }}
				{{ $service.Name }} http://{{ $service.Hostname }}:{{ $service.Port }}/_info http://{{ $service.Hostname }}:{{ $service.Port }}/{{ $service.HealthPath }}
					{{- end }}
				{{- end }}
			}
		}
	}

	# Respond 502 if service does not exists
	handle /api/* {
		respond "Bad Gateway" 502
	}

	# handle all other requests
	{{- if not (eq .Fallback "") }}
	handle {
		{{- if .Redirect }}
		redir {{ .Fallback }}
		{{- else }}
		reverse_proxy {{ .Fallback }}
		{{- end }}
		import cors
	}
	{{ end }}
}`