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

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"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).
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")

// 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)

	poolConfig := actor.GrainPoolConfig[profile.ProfileGrain]{
		MutationRegistry: reg,
		Storage:          diskStorage,
		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()

	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)

	// UCP Customer REST adapter
	auditLogPath := filepath.Join(profileDir, "audit.log")
	customerUCP := ucp.CustomerHandler(pool, auditLogPath, credStore)
	if signer := loadUCPProfileSigner(); signer != nil {
		customerUCP = ucp.WithSigning(customerUCP, signer)
		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.
	authHandler := customerauth.NewServer(credStore, pool, customerauth.NewSigner(authSecret()), 0, customerauth.Options{
		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()
	}
}