go-cart-actor/cmd/profile/main.go

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/mail"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"time"

	"git.k6n.net/mats/go-cart-actor/internal/customerauth"
	"git.k6n.net/mats/go-cart-actor/internal/ucp"
	"git.k6n.net/mats/go-cart-actor/pkg/actor"
	"git.k6n.net/mats/go-cart-actor/pkg/profile"
	"git.k6n.net/mats/go-cart-actor/pkg/proxy"
	"git.k6n.net/mats/go-cart-actor/pkg/telemetry"
	"git.k6n.net/mats/platform/config"
	"git.k6n.net/mats/platform/rabbit"
	"github.com/redis/go-redis/v9"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

// buildAuthStorage selects the credential store and login limiter that back the
// customer-auth surface. With REDIS_ADDRESS set it uses shared Redis storage so
// the service scales horizontally; otherwise it falls back to the file-backed
// store and an in-memory limiter (single-instance / local dev). A Redis failure
// at startup is fatal rather than silently degrading to per-replica state.
func buildAuthStorage(ctx context.Context, profileDir string) (customerauth.Credentials, customerauth.Limiter) {
	if addr := os.Getenv("REDIS_ADDRESS"); addr != "" {
		rdb := redis.NewClient(&redis.Options{
			Addr:     addr,
			Password: os.Getenv("REDIS_PASSWORD"),
			DB:       0,
		})
		store, err := customerauth.NewRedisCredentialStore(ctx, rdb)
		if err != nil {
			log.Fatalf("Error connecting customer-auth credential store to Redis at %s: %v", addr, err)
		}
		log.Printf("customer-auth: using shared Redis storage at %s", addr)
		return store, customerauth.NewRedisLoginLimiter(rdb, 0, 0)
	}
	credStore, err := customerauth.LoadCredentialStore(filepath.Join(profileDir, "credentials.json"))
	if err != nil {
		log.Fatalf("Error loading credential store: %v", err)
	}
	log.Print("customer-auth: REDIS_ADDRESS unset — using file credential store + in-memory limiter (single-instance only)")
	return credStore, customerauth.NewLoginLimiter(0, 0)
}

// authSecret returns the HMAC key for customer session cookies. It reads
// CUSTOMER_AUTH_SECRET; when unset it generates an ephemeral random key and
// warns that issued sessions will not survive a restart (fine for dev).
// selectAuthNotifier picks the transactional auth-message sender. When
// MAILERSEND_API_KEY is set it returns a MailerSend-backed notifier; otherwise
// it returns nil (the server defaults to LogNotifier, which is fine for dev).
func selectAuthNotifier() customerauth.Notifier {
	apiKey := os.Getenv("MAILERSEND_API_KEY")
	if apiKey == "" {
		return nil
	}
	fromAddr := strings.TrimSpace(os.Getenv("PROFILE_EMAIL_FROM_ADDRESS"))
	if fromAddr == "" {
		log.Print("warning: MAILERSEND_API_KEY set but PROFILE_EMAIL_FROM_ADDRESS is empty — falling back to LogNotifier")
		return nil
	}
	from := mail.Address{
		Name:    strings.TrimSpace(os.Getenv("PROFILE_EMAIL_FROM_NAME")),
		Address: fromAddr,
	}
	n, err := customerauth.NewMailerSendNotifier(apiKey, from)
	if err != nil {
		log.Printf("warning: mailersend notifier: %v — falling back to LogNotifier", err)
		return nil
	}
	log.Print("customer-auth: using MailerSend notifier for verification and reset emails")
	return n
}

func authSecret() []byte {
	if v := os.Getenv("CUSTOMER_AUTH_SECRET"); v != "" {
		return []byte(v)
	}
	b := make([]byte, 32)
	if _, err := rand.Read(b); err != nil {
		log.Fatalf("could not generate auth secret: %v", err)
	}
	log.Print("warning: CUSTOMER_AUTH_SECRET unset — using an ephemeral key; customer sessions will not survive a restart")
	return []byte(hex.EncodeToString(b))
}

var podIp = os.Getenv("POD_IP")
var name = os.Getenv("POD_NAME")

// profileMetrics is the shared prometheus instrumentation for the
// profile actor pool and event log. Built once at process start; the
// actor package's touch sites are nil-safe so a test binary could pass
// nil. Package-level so tests in this package can construct a pool
// without re-registering (cart/checkout use the same pattern).
var profileMetrics = actor.NewMetrics("profile")

// loadUCPProfileSigner loads the UCP ECDSA signing key from the path specified in
// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable.
func loadUCPProfileSigner() *ucp.SigningConfig {
	path := os.Getenv("UCP_SIGNING_KEY_PATH")
	if path == "" {
		return nil
	}
	pemData, err := os.ReadFile(path)
	if err != nil {
		log.Printf("ucp signing: cannot read key %s: %v", path, err)
		return nil
	}
	cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026")
	if err != nil {
		log.Printf("ucp signing: invalid key at %s: %v", path, err)
		return nil
	}
	return cfg
}

func main() {
	reg := profile.NewProfileMutationRegistry()

	profileDir := config.EnvString("PROFILE_DIR", "data/profiles")
	if err := os.MkdirAll(profileDir, 0755); err != nil {
		log.Printf("warning: could not create profile data directory %s: %v", profileDir, err)
	}
	diskStorage := actor.NewDiskStorage[profile.ProfileGrain](profileDir, reg)
	diskStorage.SetMetrics(profileMetrics)

	poolConfig := actor.GrainPoolConfig[profile.ProfileGrain]{
		MutationRegistry: reg,
		Storage:          diskStorage,
		Metrics:          profileMetrics,
		Spawn: func(ctx context.Context, id uint64) (actor.Grain[profile.ProfileGrain], error) {
			ret := profile.NewProfileGrain(id, time.Now())
			err := diskStorage.LoadEvents(ctx, id, ret)
			return ret, err
		},
		// Destroy is an optional grain-eviction cleanup hook; profiles need none
		// (disk storage persists via its own loop), so it's left unset — purge()
		// skips a nil Destroy.
		// SpawnHost makes the pool cluster-aware: when a grain is owned by a peer
		// replica, the pool proxies to it over gRPC :1337 (control) + HTTP :8080
		// (requests). With POD_IP unset and no discovered peers this is never
		// invoked, so the service still runs fine single-node.
		SpawnHost: func(host string) (actor.Host[profile.ProfileGrain], error) {
			return proxy.NewRemoteHost[profile.ProfileGrain](host)
		},
		TTL:      10 * time.Minute,
		PoolSize: 65535,
		Hostname: podIp,
	}

	pool, err := actor.NewSimpleGrainPool(poolConfig)
	if err != nil {
		log.Fatalf("Error creating profile pool: %v\n", err)
	}

	// Stream applied profile mutations to the shared mutations exchange (routing
	// key mutation.profile) for the backoffice live feed. Best-effort: disabled
	// when AMQP_URL is unset or the broker is unreachable.
	if amqpURL := os.Getenv("AMQP_URL"); amqpURL != "" {
		if conn, derr := rabbit.Dial(amqpURL, "cart-profile"); derr != nil {
			log.Printf("profile: AMQP connect failed, mutation feed disabled: %v", derr)
		} else {
			feed := actor.NewAmqpListener(conn.Connection(), "profile", actor.MutationSummary)
			feed.DefineTopics()
			pool.AddListener(feed)
			log.Printf("profile: mutation feed enabled (mutation.profile)")
		}
	}

	// Clustering control plane: gRPC server on :1337 serves peer pools'
	// ownership announcements and remote Apply/Get calls; UseDiscovery watches
	// for sibling profile pods (label actor-pool=profile) and wires them into the
	// pool. Both are no-ops in single-node mode (POD_IP unset → nil discovery).
	grpcSrv, err := actor.NewControlServer[profile.ProfileGrain](actor.DefaultServerConfig(), pool)
	if err != nil {
		log.Fatalf("Error starting control plane gRPC server: %v\n", err)
	}
	defer grpcSrv.GracefulStop()

	UseDiscovery(pool)

	mux := http.NewServeMux()
	debugMux := http.NewServeMux()

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()

	// Data Retention (C5) GDPR Purge
	retentionDays := config.EnvInt("PROFILE_RETENTION_DAYS", 0, func(n int) bool { return n >= 0 })
	if retentionDays > 0 {
		purgeInterval := config.EnvDuration("PROFILE_PURGE_INTERVAL", 24*time.Hour)
		log.Printf("profile: data retention enabled. Retaining profiles for %d days, purging every %v", retentionDays, purgeInterval)
		go func() {
			ticker := time.NewTicker(purgeInterval)
			defer ticker.Stop()

			// Run immediately on startup
			runPurge(ctx, diskStorage, pool, retentionDays)

			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					runPurge(ctx, diskStorage, pool, retentionDays)
				}
			}
		}()
	} else {
		log.Print("profile: data retention / GDPR purge disabled (PROFILE_RETENTION_DAYS not set or <= 0)")
	}

	otelShutdown, err := telemetry.SetupOTelSDK(ctx)
	if err != nil {
		log.Fatalf("Unable to start otel %v", err)
	}

	// Customer auth storage and failed-login limiter.
	// Credentials and the failed-login limiter use shared Redis storage when
	// REDIS_ADDRESS is set (so the service scales horizontally), and fall back to
	// the file store + in-memory limiter for single-instance / local dev. Session
	// and verify/reset tokens are stateless HMAC values, so a shared
	// CUSTOMER_AUTH_SECRET is all they need to work across replicas.
	credStore, limiter := buildAuthStorage(ctx, profileDir)

	// Email index: in-memory email→profileID map so the order service can
	// look up email preferences without scanning all profiles. Built from
	// the event log on disk at startup, then kept live by the UCP handlers.
	emailIndex := profile.NewProfileEmailIndex()
	stateStorage := actor.NewState(reg)
	profile.BuildEmailIndex(profileDir, stateStorage, reg, emailIndex)

	// UCP Customer REST adapter
	auditLogPath := filepath.Join(profileDir, "audit.log")
	// Session signer is shared by the auth server (HMAC verifier for the
	// "sid" cookie); NewSigner is stateless once built so a single
	// instance can be reused across the auth server's NewServer call.
	sessionSigner := customerauth.NewSigner(authSecret())

	// Cluster-aware ProfileApplier: routes Get/Apply to the replica that
	// currently owns the grain. With a remote owner we forward there so
	// the request reads the authoritative in-memory state; with no owner
	// we let pool.Get / pool.Apply spawn the grain locally from the
	// disk-backed event log and broadcast ownership. Centralising this
	// choice in the applicr (rather than HTTP middleware) prevents the
	// split-brain hazard where a non-owner pod reads a stale or empty
	// grain from disk and caches it under its own (conflicting)
	// ownership.
	applier := &clusterAwareApplier{pool: pool}

	customerUCP := ucp.CustomerHandler(applier, auditLogPath,
		ucp.WithEmailIndex(emailIndex),
		ucp.WithCredentialDeleter(credStore),
	)
	if ucpSigner := loadUCPProfileSigner(); ucpSigner != nil {
		customerUCP = ucp.WithSigning(customerUCP, ucpSigner)
		log.Print("ucp customer signing enabled")
	}
	// StripPrefix is required because the sub-mux (CustomerHandler) registers
	// routes like "GET /{id}" which expect a relative path; Go's ServeMux
	// passes the full request path without stripping the prefix.
	mux.Handle("/ucp/v1/customers/", http.StripPrefix("/ucp/v1/customers", customerUCP))

	// Customer auth: password signup/login + session cookies + identity linking.
	authNotifier := selectAuthNotifier()
	authHandler := customerauth.NewServer(credStore, applier, sessionSigner, 0, customerauth.Options{
		Notifier:             authNotifier,
		Limiter:              limiter,
		BaseURL:              os.Getenv("CUSTOMER_AUTH_BASE_URL"),
		RequireVerifiedEmail: os.Getenv("CUSTOMER_AUTH_REQUIRE_VERIFIED") == "true",
	}).Handler()
	mux.Handle("/ucp/v1/auth/", http.StripPrefix("/ucp/v1/auth", authHandler))

	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		grainCount, capacity := pool.LocalUsage()
		if grainCount >= capacity {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("grain pool at capacity"))
			return
		}
		if !pool.IsHealthy() {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("pool not healthy"))
			return
		}
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})

	mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("1.0.0"))
	})

	srv := &http.Server{
		Addr:         config.EnvString("PROFILE_ADDR", ":8080"),
		BaseContext:  func(net.Listener) context.Context { return ctx },
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 20 * time.Second,
		Handler:      otelhttp.NewHandler(mux, "/"),
	}

	defer func() {
		fmt.Println("Shutting down profile service")
		otelShutdown(context.Background())
		diskStorage.Close()
		pool.Close()
	}()

	srvErr := make(chan error, 1)
	go func() {
		srvErr <- srv.ListenAndServe()
	}()

	log.Print("Profile server started at port 8080")

	go http.ListenAndServe(":8081", debugMux)

	select {
	case err = <-srvErr:
		log.Fatalf("Unable to start server: %v", err)
	case <-ctx.Done():
		stop()
	}
}

func runPurge(ctx context.Context, diskStorage *actor.DiskStorage[profile.ProfileGrain], pool *actor.SimpleGrainPool[profile.ProfileGrain], retentionDays int) {
	log.Printf("profile: starting data retention purge...")
	retention := time.Duration(retentionDays) * 24 * time.Hour

	localIds := make(map[uint64]bool)
	for _, id := range pool.GetLocalIds() {
		localIds[id] = true
	}
	isGrainActive := func(id uint64) bool {
		return localIds[id]
	}

	purged, err := diskStorage.PurgeOlderThan(ctx, retention, isGrainActive)
	if err != nil {
		log.Printf("profile: data retention purge failed: %v", err)
	} else {
		log.Printf("profile: data retention purge completed. Purged %d expired profile logs", purged)
	}
}