mpc/cmd/mpcd/main.go

package main

import (
	"context"
	"crypto/ed25519"
	crypto_elliptic "crypto/elliptic"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"golang.org/x/crypto/ripemd160"
	"golang.org/x/crypto/sha3"

	"github.com/mr-tron/base58"

	"github.com/luxfi/mpc/pkg/address"

	"github.com/nats-io/nats.go"
	"github.com/spf13/viper"
	"github.com/urfave/cli/v3"

	"github.com/hanzoai/base"
	"github.com/hanzoai/base/apis"
	"github.com/hanzoai/base/core"
	"github.com/luxfi/hsm"

	uimpc "github.com/luxfi/mpc/ui"

	"github.com/luxfi/threshold/pkg/thresholdd"

	mpcapi "github.com/luxfi/mpc/pkg/api"
	"github.com/luxfi/mpc/pkg/backup"
	"github.com/luxfi/mpc/pkg/db"
	"github.com/luxfi/mpc/pkg/event"
	"github.com/luxfi/mpc/pkg/eventconsumer"
	"github.com/luxfi/mpc/pkg/keyinfo"
	"github.com/luxfi/mpc/pkg/kvstore"
	"github.com/luxfi/mpc/pkg/logger"
	"github.com/luxfi/mpc/pkg/messaging"
	"github.com/luxfi/mpc/pkg/mpc"
	"github.com/luxfi/mpc/pkg/transport"
	"github.com/luxfi/mpc/pkg/types"
)

const (
	Version                    = "0.4.0"
	DefaultBackupPeriodSeconds = 300 // (5 minutes)
)

func main() {
	app := &cli.Command{
		Name:    "mpcd",
		Usage:   "MPC daemon for threshold signatures (consensus-embedded)",
		Version: Version,
		Commands: []*cli.Command{
			{
				Name:  "start",
				Usage: "Start an MPC node",
				Flags: []cli.Flag{
					&cli.StringFlag{
						Name:  "node-id",
						Usage: "Node ID",
					},
					&cli.StringFlag{
						Name:  "listen",
						Usage: "P2P ZAP listen address",
						Value: ":9999",
					},
					&cli.StringFlag{
						Name:  "api",
						Usage: "Internal API listen address",
						Value: ":9800",
					},
					&cli.StringFlag{
						Name:  "data",
						Usage: "Data directory",
					},
					&cli.StringFlag{
						Name:  "keys",
						Usage: "Keys directory",
					},
					&cli.IntFlag{
						Name:    "threshold",
						Aliases: []string{"t"},
						Usage:   "Signing threshold",
						Value:   2,
					},
					&cli.StringSliceFlag{
						Name:  "peer",
						Usage: "Peer address (can be specified multiple times)",
					},
					&cli.StringFlag{
						Name:  "log-level",
						Usage: "Log level (debug, info, warn, error)",
						Value: "info",
					},
					&cli.StringFlag{
						Name:  "api-listen",
						Usage: "Dashboard API listen address",
						Value: ":8081",
					},
					&cli.StringFlag{
						Name:  "jwt-secret",
						Usage: "JWT signing secret for dashboard auth",
					},
					&cli.StringFlag{
						Name:    "kms-zap-listen",
						Usage:   "KMS-facing ZAP server listen address (luxfi/kms dials this for threshold ops). Empty disables. Default :9653 matches the operator zapPort.",
						Sources: cli.EnvVars("MPC_KMS_ZAP_LISTEN"),
						Value:   ":9653",
					},
					&cli.StringFlag{
						Name:    "threshold-listen",
						Usage:   "Embedded threshold ZAP dispatcher listen address (luxfi/threshold/pkg/thresholdd surface: cggmp21/frost/pulsar/corona/magnetar/doerner). Empty disables. Default 127.0.0.1:7301 — process-local IPC; production fronts this via the cluster ingress + KMS.",
						Sources: cli.EnvVars("MPC_THRESHOLD_LISTEN"),
						Value:   "127.0.0.1:7301",
					},
					&cli.StringFlag{
						Name:    "chain-profile",
						Usage:   "Node chain-security posture for the embedded threshold dispatcher's strict-PQ gate: strict-pq|legacy-compat|unknown. Default strict-pq (fail-closed) — the single-party dealer shortcut on ctx-bound sign_ctx (pulsar/magnetar) is REFUSED unless a chain is allow-listed via --legacy-chains. Set legacy-compat only for dev / non-PQ chains.",
						Sources: cli.EnvVars("MPC_CHAIN_PROFILE"),
						Value:   "strict-pq",
					},
					&cli.StringFlag{
						Name:    "legacy-chains",
						Usage:   "Comma/space-separated chainIDs exempt from the strict-PQ gate (permitted the dealer shortcut) even when --chain-profile is strict-pq. Empty by default — nothing is exempt.",
						Sources: cli.EnvVars("MPC_LEGACY_CHAINS"),
					},
					// HSM / password provider flags
					&cli.StringFlag{
						Name:    "hsm-provider",
						Usage:   "Password provider type: aws|gcp|azure|env|file (default: env)",
						Sources: cli.EnvVars("MPC_HSM_PROVIDER"),
						Value:   "env",
					},
					&cli.StringFlag{
						Name:    "hsm-key-id",
						Usage:   "HSM key ARN/name/path for ZapDB password decryption",
						Sources: cli.EnvVars("MPC_HSM_KEY_ID"),
					},
					// HSM signer flags (for co-signing)
					&cli.StringFlag{
						Name:    "hsm-signer",
						Usage:   "Signer provider for intent co-signing: aws|gcp|azure|zymbit|mldsa|local (default: local)",
						Sources: cli.EnvVars("MPC_HSM_SIGNER"),
						Value:   "local",
					},
					&cli.StringFlag{
						Name:    "hsm-signer-key-id",
						Usage:   "HSM signer key ARN/name for co-signing operations",
						Sources: cli.EnvVars("MPC_HSM_SIGNER_KEY_ID"),
					},
					&cli.BoolFlag{
						Name:    "hsm-attest",
						Usage:   "Enable HSM attestation on threshold signature shares (binds shares to hardware)",
						Sources: cli.EnvVars("MPC_HSM_ATTEST"),
						Value:   false,
					},
					&cli.BoolFlag{
						Name:  "debug",
						Usage: "Enable debug logging",
						Value: false,
					},
				},
				Action: func(ctx context.Context, c *cli.Command) error {
					return runNodeConsensus(ctx, c)
				},
			},
			{
				Name:  "version",
				Usage: "Display detailed version information",
				Action: func(ctx context.Context, c *cli.Command) error {
					fmt.Printf("mpcd version %s\n", Version)
					return nil
				},
			},
		},
	}

	if err := app.Run(context.Background(), os.Args); err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
}

// checkRequiredConfigValues verifies required viper config values are present.
// skipPasswordCheck should be true when ZapDB password will be resolved via HSM provider.
func checkRequiredConfigValues(skipPasswordCheck bool) {
	// Show warning if we're using file-based config but no password is set
	if !skipPasswordCheck && viper.GetString("zapdb_password") == "" {
		logger.Fatal("ZapDB password is required", nil)
	}

	if viper.GetString("event_initiator_pubkey") == "" {
		logger.Fatal("Event initiator public key is required", nil)
	}
}

// resolveZapDBPassword returns the ZapDB encryption password using the HSM
// password provider infrastructure. Provider type comes from --hsm-provider
// (or MPC_HSM_PROVIDER env var), defaulting to "env" for backward compat.
// When a cloud provider (aws/gcp/azure) is configured, the password is
// decrypted from ZAPDB_ENCRYPTED_PASSWORD via the cloud KMS at startup.
func resolveZapDBPassword(ctx context.Context, c *cli.Command) string {
	hsmProviderType := c.String("hsm-provider")
	hsmKeyID := c.String("hsm-key-id")
	environment := viper.GetString("environment")

	// In production, env/file password providers are rejected. ZapDB passwords
	// must come from a cloud HSM (aws/gcp/azure) so they aren't readable via kubectl exec.
	if environment == "production" && (hsmProviderType == "" || hsmProviderType == "env" || hsmProviderType == "file") {
		logger.Fatal("Production requires cloud HSM password provider (aws/gcp/azure); env/file providers are not permitted",
			fmt.Errorf("MPC_HSM_PROVIDER=%q", hsmProviderType))
	}

	provider, err := hsm.NewPasswordProvider(hsmProviderType, nil)
	if err != nil {
		logger.Fatal("Failed to create HSM password provider", fmt.Errorf("provider=%s: %w", hsmProviderType, err))
	}

	password, err := provider.GetPassword(ctx, hsmKeyID)
	if err != nil {
		if environment == "production" {
			logger.Fatal("HSM provider failed in production; cannot fall back to config",
				fmt.Errorf("provider=%s: %w", hsmProviderType, err))
		}
		// Fall back to viper config for dev/staging only
		password = viper.GetString("zapdb_password")
		if password == "" {
			logger.Fatal("ZapDB password is required: HSM provider failed and no zapdb_password in config",
				fmt.Errorf("provider=%s: %w", hsmProviderType, err))
		}
		logger.Warn("ZapDB password loaded from config (non-production only)", "environment", environment)
		return password
	}

	logger.Info("ZapDB password loaded via HSM provider", "provider", hsmProviderType)
	return password
}

func NewZapKV(nodeName, nodeID, password string) *kvstore.Store {
	// ZapDB KV store
	// Use configured db_path or default to current directory + "db"
	basePath := viper.GetString("db_path")
	if basePath == "" {
		basePath = filepath.Join(".", "db")
	}
	dbPath := filepath.Join(basePath, nodeName)

	// Use configured backup_dir or default to current directory + "backups"
	backupDir := viper.GetString("backup_dir")
	if backupDir == "" {
		backupDir = filepath.Join(".", "backups")
	}

	// Create ZapDB config
	config := kvstore.Config{
		NodeID:    nodeName,
		Key:       []byte(password),
		BackupKey: []byte(password), // Using same key for backup encryption
		Dir:       backupDir,
		Path:      dbPath,
	}

	kv, err := kvstore.New(config)
	if err != nil {
		logger.Fatal("Failed to create zapdb store", err)
	}
	logger.Info("Connected to zapdb store", "path", dbPath, "backup_dir", backupDir)
	return kv
}

func StartPeriodicBackup(ctx context.Context, zapKV *kvstore.Store, periodSeconds int) func() {
	if periodSeconds <= 0 {
		periodSeconds = DefaultBackupPeriodSeconds
	}
	backupTicker := time.NewTicker(time.Duration(periodSeconds) * time.Second)
	backupCtx, backupCancel := context.WithCancel(ctx)
	go func() {
		for {
			select {
			case <-backupCtx.Done():
				logger.Info("Backup background job stopped")
				return
			case <-backupTicker.C:
				logger.Info("Running periodic ZapDB backup...")
				err := zapKV.Backup()
				if err != nil {
					logger.Error("Periodic ZapDB backup failed", err)
				} else {
					logger.Info("Periodic ZapDB backup completed successfully")
				}
			}
		}
	}()
	return backupCancel
}

// runNodeConsensus runs the MPC node with consensus-embedded transport
// keygenDegreeForThreshold maps the operator-facing --threshold (the number of
// signers required, N in "N-of-M") to the CGGMP21 polynomial degree fed to
// cmp.Keygen (which needs degree+1 signers). degree = threshold-1, floored at
// 0. threshold<=1 → degree 0 (1-of-n, dev/single-signer only). Keeping this a
// named, tested function makes the security-critical off-by-one explicit.
func keygenDegreeForThreshold(threshold int) int {
	if threshold < 1 {
		return 0
	}
	return threshold - 1
}

// isSelfAddr reports whether a --peer address names THIS node.
//
// Two forms count, and neither is a guess:
//   - the address equals the node's own --listen (identical spelling), or
//   - its host's FIRST DNS LABEL equals the node id. That is the k8s
//     StatefulSet shape and the only one this fleet uses:
//     `mpc-node-0.mpc-node-headless.hanzo-mpc.svc.cluster.local:9999` on the
//     pod whose id is `mpc-node-0`. The label is compared whole, so
//     `mpc-node-10` never matches `mpc-node-1`.
//
// Anything else is a peer. A false negative merely restores the old
// double-count; a false positive would drop a real peer, so the match is
// deliberately exact rather than a prefix test.
func isSelfAddr(addr, nodeID, listenAddr string) bool {
	addr = strings.TrimSpace(addr)
	if addr == "" || nodeID == "" {
		return false
	}
	if addr == listenAddr {
		return true
	}
	host := addr
	if h, _, err := net.SplitHostPort(addr); err == nil {
		host = h
	}
	label := host
	if i := strings.IndexByte(host, '.'); i >= 0 {
		label = host[:i]
	}
	return label == nodeID
}

// peerIDFromAddr derives a peer's node id from its ADDRESS, which is the only
// form an unprefixed `--peer` carries.
//
// Every party must agree on the NAME of every other party, because the keygen
// handler admits a message only from an id in its own party set. A positional
// `peer-N` cannot match a real node id by construction, so a fleet whose flags
// are unprefixed builds THREE DIFFERENT party sets — mpc-node-1 called node-2
// "peer-2" while node-2 called itself "mpc-node-2" — and every round-2 message
// was refused by every node. Measured on hanzo-mpc: partyIDs
// ["mpc-node-0:keygen:1","mpc-node-1:keygen:1","peer-2:keygen:1"], zero messages
// accepted, keygen timing out at 300s, and each crypto deposit answering 503
// "the custody service is not accepting new addresses".
//
// The first DNS label IS the node id under StatefulSet naming
// (mpc-node-2.mpc-node-headless.ns.svc.cluster.local -> mpc-node-2), which is
// exactly the rule isSelfAddr already uses to recognise self. Deriving both
// sides the same way is what makes the sets agree.
//
// A bare IP has no label to read, so it keeps the positional id: the caller
// must then use the `nodeID@host:port` form, which states the name outright.
func peerIDFromAddr(addr string, index int) string {
	host := strings.TrimSpace(addr)
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}
	label := host
	if i := strings.IndexByte(host, '.'); i >= 0 {
		label = host[:i]
	}
	if label == "" || net.ParseIP(host) != nil {
		return fmt.Sprintf("peer-%d", index)
	}
	return label
}

func runNodeConsensus(ctx context.Context, c *cli.Command) error {
	nodeID := c.String("node-id")
	listenAddr := c.String("listen")
	dataDir := c.String("data")
	keysDir := c.String("keys")
	threshold := c.Int("threshold")
	peers := c.StringSlice("peer")
	logLevel := c.String("log-level")
	debug := c.Bool("debug")

	// --threshold is the number of signers required; it drives the
	// HasSigningQuorum health gate (readyCount >= threshold). The CGGMP21
	// keygen polynomial DEGREE, however, is read separately from viper key
	// "mpc_threshold" (pkg/eventconsumer.NewEventConsumer →
	// CreateKeyGenSession → cmp.Keygen). NOTHING wired the flag into that key,
	// and mpcd never calls config.InitViperConfig (no AutomaticEnv), so
	// viper.GetInt("mpc_threshold") was always 0 → every wallet was keyed at
	// degree 0 = 1-of-n: NO threshold security, a single share signs. Bridge
	// them here. cmp requires degree+1 signers, so degree = threshold-1:
	// threshold=3 → degree 2 → 3-of-5; threshold=2 → degree 1 → 2-of-3.
	// viper.Set has the highest precedence, so this wins without a config file.
	keygenDegree := keygenDegreeForThreshold(threshold)
	viper.Set("mpc_threshold", keygenDegree)

	if nodeID == "" {
		return fmt.Errorf("--node-id is required in consensus mode")
	}
	if dataDir == "" {
		return fmt.Errorf("--data is required in consensus mode")
	}

	// Initialize logger
	logger.Init("consensus", debug || logLevel == "debug")
	logger.Info("Starting MPC node in consensus mode",
		"nodeID", nodeID,
		"listen", listenAddr,
		"dataDir", dataDir,
		"signersRequired", threshold,
		"keygenDegree", keygenDegree,
		"peers", len(peers),
	)

	// Ensure directories exist
	if err := os.MkdirAll(dataDir, 0750); err != nil {
		return fmt.Errorf("failed to create data directory: %w", err)
	}
	if keysDir == "" {
		keysDir = filepath.Join(dataDir, "keys")
	}
	if err := os.MkdirAll(keysDir, 0700); err != nil {
		return fmt.Errorf("failed to create keys directory: %w", err)
	}

	// Load or generate identity
	privKey, pubKey, err := loadOrGenerateIdentity(keysDir, nodeID)
	if err != nil {
		return fmt.Errorf("failed to load/generate identity: %w", err)
	}

	// Create consensus identity store for verifying messages
	consensusIdentity := NewConsensusIdentityStore(nodeID, privKey, pubKey)

	// Build peer map. Self is in it deliberately (the transport's address book
	// needs its own address), keyed by the REAL node id — every other entry is a
	// peer.
	//
	// A --peer that IS this node is dropped, and that is what makes the ordinary
	// k8s deployment correct: a StatefulSet renders ONE command for every
	// replica, so the natural spelling passes the whole ensemble to each node
	// (`--peer=mpc-node-0… --peer=mpc-node-1… --peer=mpc-node-2…`) and each one
	// receives itself. Without this drop, self is counted TWICE — once as
	// `nodeID`, once as a synthetic `peer-N` — so a 3-node fleet reports 4
	// expected members and can never satisfy the all-ready equality that gates
	// KEYGEN. Measured on hanzo-mpc: `ready=5 expected=4`, forever, with
	// `signing_quorum: true` beside it (signing uses >=, so only keygen was
	// dead) — every crypto deposit-address request answered 503 "peers not
	// ready" while the fleet looked healthy.
	//
	// Self is recognised by ADDRESS, because that is the only form the flag
	// carries here: an unprefixed --peer gets a positional `peer-N` id that
	// cannot match a real node id by construction.
	peerMap := make(map[string]string)
	peerMap[nodeID] = listenAddr
	for i, peer := range peers {
		// Parse peer address - format: "nodeID@host:port" or just "host:port"
		parts := strings.SplitN(peer, "@", 2)
		if len(parts) == 2 {
			if parts[0] == nodeID || isSelfAddr(parts[1], nodeID, listenAddr) {
				continue
			}
			peerMap[parts[0]] = parts[1]
			continue
		}
		if isSelfAddr(peer, nodeID, listenAddr) {
			continue
		}
		peerMap[peerIDFromAddr(peer, i)] = peer
	}

	// Get ZapDB password via HSM provider (supports AWS KMS, GCP Cloud KMS, Azure Key Vault, env, file)
	zapDBPassword := resolveZapDBPassword(ctx, c)

	// Create transport factory (uses ZapDB for embedded key-share storage).
	// OnPeerIdentity feeds peer Ed25519 public keys learned from the transport
	// identity handshake into the consensus identity store, so wire-message
	// signature verification (keygen/signing rounds) can resolve peer keys.
	// This is the consensus-mode analog of the legacy NATS path's peers.json +
	// *_identity.json seeding; without it every inbound peer message is dropped.
	factoryCfg := transport.FactoryConfig{
		NodeID:        nodeID,
		ListenAddr:    listenAddr,
		Peers:         peerMap,
		PrivateKey:    privKey,
		PublicKey:     pubKey,
		ZapDBPath:     filepath.Join(dataDir, "db"),
		ZapDBPassword: zapDBPassword,
		BackupDir:     filepath.Join(dataDir, "backups"),
		OnPeerIdentity: func(peerNodeID string, peerPubKey ed25519.PublicKey) {
			consensusIdentity.AddPeerPublicKey(peerNodeID, peerPubKey)
		},
	}

	factory, err := transport.NewFactory(factoryCfg)
	if err != nil {
		return fmt.Errorf("failed to create transport factory: %w", err)
	}

	// Start transport
	if err := factory.Start(ctx); err != nil {
		return fmt.Errorf("failed to start transport: %w", err)
	}
	defer factory.Stop()

	// Create MPC node with consensus transport
	peerIDs := make([]string, 0, len(peerMap)-1)
	for id := range peerMap {
		if id != nodeID {
			peerIDs = append(peerIDs, id)
		}
	}

	// Create PubSub adapter for messaging
	pubSub := NewConsensusPubSubAdapter(factory.PubSub())

	// Create message queue adapters
	genKeyResultQueue := NewConsensusMessageQueue(factory.Transport(), nodeID, "keygen")
	signingResultQueue := NewConsensusMessageQueue(factory.Transport(), nodeID, "signing")
	reshareResultQueue := NewConsensusMessageQueue(factory.Transport(), nodeID, "reshare")

	logger.Info("Node is running in consensus mode", "nodeID", nodeID)

	// Create peer registry using consensus membership
	peerRegistry := NewConsensusPeerRegistry(factory.Registry(), nodeID, peerIDs)

	// Create MPC node
	mpcNode := mpc.NewNode(
		nodeID,
		peerIDs,
		pubSub,
		factory.KVStore(),
		NewConsensusKeyInfoStore(factory.KeyInfoStore(), peerRegistry),
		peerRegistry,
		consensusIdentity,
	)

	// Create event consumer
	eventConsumer := eventconsumer.NewEventConsumer(
		mpcNode,
		pubSub,
		genKeyResultQueue,
		signingResultQueue,
		reshareResultQueue,
		consensusIdentity,
	)
	eventConsumer.Run()
	defer eventConsumer.Close()

	// Mark as ready
	if err := peerRegistry.Ready(); err != nil {
		logger.Error("Failed to mark peer registry as ready", err)
	}
	logger.Info("[READY] Node is ready (consensus mode)", "nodeID", nodeID)

	// MPCBackend wraps the consensus transport so the internal HTTP API
	// (port 9800), the Dashboard API (port 8081), and the KMS ZAP server
	// (port 9653) can all drive keygen/sign/reshare against the same node.
	// Hoisted above the internal HTTP API block so the /sign handler can call
	// TriggerSign; it depends only on values constructed above (pubSub,
	// peerRegistry, factory, keyInfoStore, identity, nodeID, threshold).
	mpcBackend := &ConsensusMPCBackend{
		pubSub:       pubSub,
		peerRegistry: peerRegistry,
		factory:      factory,
		keyInfoStore: factory.KeyInfoStore(),
		identity:     consensusIdentity,
		nodeID:       nodeID,
		threshold:    threshold,
	}

	// Per-node idempotency cache for the internal /sign endpoint. Anti-oracle:
	// a reused idempotency key with conflicting content is refused without
	// signing; concurrent duplicates collapse to a single threshold-sign.
	signIdem := newSignIdempotencyCache()

	// Start HTTP API server (internal MPC node API on port 9800)
	apiAddr := c.String("api")
	if apiAddr != "" {
		// Internal API bearer token — required for all endpoints except /health.
		// Source: MPC_INTERNAL_API_KEY env var. In production the StatefulSet
		// injects this from the KMS-synced mpc-secrets K8s Secret.
		internalAPIKey := os.Getenv("MPC_INTERNAL_API_KEY")
		if internalAPIKey == "" {
			// Derive a deterministic key from the node's Ed25519 private key so
			// all nodes in the cluster share the same key without extra config.
			// SHA-256(privKey || "mpc-internal-api") truncated to hex.
			h := sha256.Sum256(append(privKey.Seed(), []byte("mpc-internal-api")...))
			internalAPIKey = hex.EncodeToString(h[:])
			logger.Warn("MPC_INTERNAL_API_KEY not set; derived internal API key from node identity (set MPC_INTERNAL_API_KEY in production)")
		}

		// Rate limiter: 10 requests/min for mutating endpoints (keygen, backup).
		internalRL := mpcapi.NewRateLimiter(10)

		// internalAuth is middleware that gates all mutating internal endpoints
		// behind a bearer token. /health is exempt (K8s probes need it).
		internalAuth := func(next http.HandlerFunc) http.HandlerFunc {
			return func(w http.ResponseWriter, r *http.Request) {
				auth := r.Header.Get("Authorization")
				if auth == "" || auth != "Bearer "+internalAPIKey {
					w.Header().Set("Content-Type", "application/json")
					w.WriteHeader(http.StatusUnauthorized)
					json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"})
					return
				}
				next.ServeHTTP(w, r)
			}
		}

		// internalRateLimit wraps a handler with the tight rate limiter.
		internalRateLimit := func(next http.HandlerFunc) http.HandlerFunc {
			return func(w http.ResponseWriter, r *http.Request) {
				ip := r.RemoteAddr
				if host, _, err := net.SplitHostPort(ip); err == nil {
					ip = host
				}
				if !internalRL.Allow(ip) {
					w.Header().Set("Content-Type", "application/json")
					w.WriteHeader(http.StatusTooManyRequests)
					json.NewEncoder(w).Encode(map[string]string{"error": "rate limit exceeded"})
					return
				}
				next.ServeHTTP(w, r)
			}
		}

		mux := http.NewServeMux()
		// Health probe handler — unauthenticated (K8s liveness/readiness probes).
		// Served on both /health (legacy) and /healthz (platform standard).
		//
		// Red finding P0-1 (2026-04-20): the previous implementation returned
		// 503 whenever ArePeersReady() was false — i.e. whenever ANY peer was
		// unreachable. For a 3-node 2-of-3 ensemble that collapsed availability
		// the moment one peer restarted, because the surviving two nodes
		// reported themselves "unhealthy" even though they could still form a
		// 2-of-3 signing quorum between them.
		//
		// Fixed behaviour: the node is healthy when it has enough ready peers
		// (including self) to form a t-of-n signing quorum using the
		// operator-facing threshold (CRD `threshold` / mpcd --threshold flag
		// = minimum signers required). For a 2-of-3 ensemble (threshold=2),
		// quorum holds when 2 or more nodes are ready. See
		// HasSigningQuorum in pkg/transport/registry.go for the arithmetic.
		//
		// Keygen of fresh wallets still requires all peers (checked separately
		// on the /keygen endpoint via peerRegistry.ArePeersReady()). Signing
		// with an existing key only needs threshold parties, which is what
		// /healthz now reports.
		healthHandler := func(w http.ResponseWriter, r *http.Request) {
			// allReady: every expected peer connected. Used for observability.
			allReady := peerRegistry.ArePeersReady()
			// quorum: enough ready peers for signing. Gates 200 vs 503.
			quorum := peerRegistry.HasSigningQuorum(threshold)
			readyCount := peerRegistry.GetReadyPeersCount()
			connected := factory.Transport().GetPeers()

			status := "healthy"
			httpCode := http.StatusOK
			switch {
			case !quorum:
				status = "degraded"
				httpCode = http.StatusServiceUnavailable
			case !allReady:
				// Signing quorum holds, but we're missing non-critical peers.
				// Stay 200 — the node is usable — but report the nuance.
				status = "healthy-reduced"
			}

			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(httpCode)
			resp := map[string]interface{}{
				"status":          status,
				"node_id":         nodeID,
				"mode":            "consensus",
				"expected_peers":  len(peerIDs),
				"connected_peers": len(connected),
				"ready":           allReady,
				"signing_quorum":  quorum,
				"ready_count":     readyCount,
				"threshold":       threshold,
				// threshold = minimum signers required (operator surface).
				// Present alongside ready_count so an operator can see how
				// close to losing quorum the ensemble is.
				"version": Version,
			}
			json.NewEncoder(w).Encode(resp)
		}
		mux.HandleFunc("/healthz", healthHandler)
		mux.HandleFunc("/keys", internalAuth(func(w http.ResponseWriter, r *http.Request) {
			keys, err := factory.KeyInfoStore().ListKeys()
			w.Header().Set("Content-Type", "application/json")
			if err != nil {
				w.WriteHeader(http.StatusInternalServerError)
				json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
				return
			}
			json.NewEncoder(w).Encode(keys)
		}))
		mux.HandleFunc("/backup", internalAuth(internalRateLimit(func(w http.ResponseWriter, r *http.Request) {
			if r.Method != http.MethodPost {
				w.WriteHeader(http.StatusMethodNotAllowed)
				return
			}
			w.Header().Set("Content-Type", "application/json")
			logger.Info("Audit: backup triggered", "nodeID", nodeID, "remote", r.RemoteAddr)
			if zapKV, ok := factory.KVStore().(*kvstore.Store); ok && zapKV.Exec != nil {
				s3Cfg := backup.S3ConfigFromEnv(nodeID)
				mgr, err := backup.NewManager(zapKV.Exec, filepath.Join(dataDir, "backups"), nodeID, 0, s3Cfg)
				if err != nil {
					w.WriteHeader(http.StatusInternalServerError)
					json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
					return
				}
				if err := mgr.RunBackup(); err != nil {
					w.WriteHeader(http.StatusInternalServerError)
					json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
					return
				}
				json.NewEncoder(w).Encode(map[string]string{"status": "backup completed"})
			} else {
				w.WriteHeader(http.StatusServiceUnavailable)
				json.NewEncoder(w).Encode(map[string]string{"error": "backup not available"})
			}
		})))
		mux.HandleFunc("/keygen", internalAuth(internalRateLimit(func(w http.ResponseWriter, r *http.Request) {
			if r.Method != http.MethodPost {
				w.WriteHeader(http.StatusMethodNotAllowed)
				return
			}
			w.Header().Set("Content-Type", "application/json")

			if !peerRegistry.ArePeersReady() {
				w.WriteHeader(http.StatusServiceUnavailable)
				json.NewEncoder(w).Encode(map[string]string{"error": "peers not ready"})
				return
			}

			// Parse request body — orgID is required for tenant isolation.
			var req struct {
				OrgID    string `json:"org_id"`
				WalletID string `json:"wallet_id"`
			}
			if r.Body != nil {
				json.NewDecoder(r.Body).Decode(&req)
			}
			if req.OrgID == "" {
				w.WriteHeader(http.StatusBadRequest)
				json.NewEncoder(w).Encode(map[string]string{"error": "org_id is required"})
				return
			}
			if req.WalletID == "" {
				// Generate a deterministic wallet ID from timestamp + node
				h := sha256.Sum256([]byte(fmt.Sprintf("%s-%d", nodeID, time.Now().UnixNano())))
				req.WalletID = hex.EncodeToString(h[:16])
			}

			walletID := req.WalletID

			// Subscribe to the result topic before triggering keygen
			resultTopic := fmt.Sprintf("mpc.mpc_keygen_result.%s", walletID)
			resultCh := make(chan *event.KeygenResultEvent, 1)
			unsub, err := pubSub.Subscribe(resultTopic, func(natMsg *nats.Msg) {
				var result event.KeygenResultEvent
				if err := json.Unmarshal(natMsg.Data, &result); err == nil {
					select {
					case resultCh <- &result:
					default:
					}
				}
			})
			if err != nil {
				w.WriteHeader(http.StatusInternalServerError)
				json.NewEncoder(w).Encode(map[string]string{"error": "failed to subscribe to result topic"})
				return
			}
			defer unsub.Unsubscribe()

			// Create and publish GenerateKeyMessage, signed by this node
			sig := consensusIdentity.SignMessage([]byte(walletID))
			msg := types.GenerateKeyMessage{
				OrgID:     req.OrgID,
				WalletID:  walletID,
				Signature: sig,
			}
			msgData, _ := json.Marshal(msg)

			if err := pubSub.Publish("mpc:generate", msgData); err != nil {
				w.WriteHeader(http.StatusInternalServerError)
				json.NewEncoder(w).Encode(map[string]string{"error": fmt.Sprintf("failed to publish keygen: %v", err)})
				return
			}

			logger.Info("Audit: keygen triggered", "nodeID", nodeID, "orgID", req.OrgID, "walletID", walletID, "remote", r.RemoteAddr)

			// Wait for result with 60s timeout
			select {
			case result := <-resultCh:
				resp := map[string]interface{}{
					"wallet_id":   result.WalletID,
					"result_type": result.ResultType,
				}
				if result.ResultType == event.ResultTypeSuccess {
					resp["ecdsa_pub_key"] = hex.EncodeToString(result.ECDSAPubKey)
					resp["eddsa_pub_key"] = hex.EncodeToString(result.EDDSAPubKey)
					if len(result.ECDSAPubKey) >= 32 {
						resp["evm_address"] = pubKeyToEVMAddress(result.ECDSAPubKey)
						resp["btc_address"] = pubKeyToBtcAddress(result.ECDSAPubKey)
					}
					if solAddr := eddsaPubKeyToSolAddress(result.EDDSAPubKey); solAddr != "" {
						resp["sol_address"] = solAddr
					}
				} else {
					resp["error"] = result.ErrorReason
					resp["error_code"] = result.ErrorCode
				}
				json.NewEncoder(w).Encode(resp)
			case <-time.After(60 * time.Second):
				w.WriteHeader(http.StatusGatewayTimeout)
				json.NewEncoder(w).Encode(map[string]string{
					"error":     "keygen timed out after 60s",
					"wallet_id": walletID,
				})
			}
		})))
		// POST /sign — threshold-sign a payload hash for an org-scoped wallet.
		// Wire contract (lux/wallet apps/backend/internal/custody/mpc.go):
		//   req:  {org_id, wallet_id, key_type, chain_id, payload_hash(hex), idempotency_key}
		//   resp: {session_id, signature(hex), status, error}
		//
		// Unlike /keygen (which needs ALL peers), signing an existing key only
		// needs a t-of-n quorum — gated on HasSigningQuorum, matching /healthz.
		// Idempotency (anti-oracle) is enforced before any signing machinery is
		// touched: empty key → 400, reused key with conflicting content → 409
		// (never signs), duplicate of an in-flight/completed request → cached.
		mux.HandleFunc("/sign", internalAuth(internalRateLimit(func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("Content-Type", "application/json")
			if r.Method != http.MethodPost {
				w.WriteHeader(http.StatusMethodNotAllowed)
				json.NewEncoder(w).Encode(map[string]string{"error": "method not allowed"})
				return
			}

			// Signing needs only a t-of-n quorum, not every peer. 503 → the
			// custody adapter surfaces "MPC peers not ready" and retries.
			if !peerRegistry.HasSigningQuorum(threshold) {
				w.WriteHeader(http.StatusServiceUnavailable)
				json.NewEncoder(w).Encode(map[string]string{"error": "peers not ready"})
				return
			}

			var req struct {
				OrgID          string `json:"org_id"`
				WalletID       string `json:"wallet_id"`
				KeyType        string `json:"key_type"`
				ChainID        int    `json:"chain_id"`
				PayloadHash    string `json:"payload_hash"`
				IdempotencyKey string `json:"idempotency_key"`
			}
			if r.Body != nil {
				if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
					w.WriteHeader(http.StatusBadRequest)
					json.NewEncoder(w).Encode(map[string]string{"error": "invalid JSON body"})
					return
				}
			}

			fields := signFields{
				OrgID:       req.OrgID,
				WalletID:    req.WalletID,
				KeyType:     req.KeyType,
				ChainID:     req.ChainID,
				PayloadHash: req.PayloadHash,
			}
			// Boundary validation: idempotency_key + org_id required.
			if err := validateSignRequest(req.IdempotencyKey, fields); err != nil {
				w.WriteHeader(http.StatusBadRequest)
				json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
				return
			}

			// payload_hash must be valid hex and exactly 32 bytes (the digest
			// to sign). Decode here so a malformed hash fails at the boundary,
			// never inside the signer.
			payload, err := hex.DecodeString(strings.TrimPrefix(req.PayloadHash, "0x"))
			if err != nil || len(payload) != 32 {
				w.WriteHeader(http.StatusBadRequest)
				json.NewEncoder(w).Encode(map[string]string{"error": "payload_hash must be 32-byte hex"})
				return
			}

			// Idempotent single-flight threshold sign. Conflict → 409, no sign.
			res, err := signIdem.Do(req.IdempotencyKey, fields, payload, mpcBackend.TriggerSign)
			if errors.Is(err, errIdempotencyConflict) {
				w.WriteHeader(http.StatusConflict)
				json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
				return
			}
			if err != nil {
				// Sign protocol failure (timeout, MPC error). Per the adapter
				// contract, 200 + non-empty error → "sign rejected". 503 is
				// reserved for no-quorum (handled above), so do NOT use it here.
				logger.Info("Audit: sign rejected", "nodeID", nodeID, "orgID", req.OrgID, "walletID", req.WalletID, "remote", r.RemoteAddr, "reason", err.Error())
				json.NewEncoder(w).Encode(map[string]string{"status": "rejected", "error": err.Error()})
				return
			}

			logger.Info("Audit: sign completed", "nodeID", nodeID, "orgID", req.OrgID, "walletID", req.WalletID, "sessionID", res.SessionID, "remote", r.RemoteAddr)
			json.NewEncoder(w).Encode(map[string]string{
				"session_id": res.SessionID,
				"signature":  res.Signature,
				"status":     res.Status,
			})
		})))

		srv := &http.Server{
			Addr:              apiAddr,
			Handler:           http.MaxBytesHandler(mux, 1<<20), // 1 MB body limit
			ReadTimeout:       30 * time.Second,
			ReadHeaderTimeout: 10 * time.Second,
			WriteTimeout:      90 * time.Second, // keygen can take 60s
			IdleTimeout:       120 * time.Second,
		}
		go func() {
			logger.Info("HTTP API server starting", "addr", apiAddr)
			if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
				logger.Error("HTTP API server failed", err)
			}
		}()
		defer srv.Close()
	}

	// Start periodic backup with optional S3 upload
	backupDir := filepath.Join(dataDir, "backups")
	if zapKV, ok := factory.KVStore().(*kvstore.Store); ok && zapKV.Exec != nil {
		s3Cfg := backup.S3ConfigFromEnv(nodeID)
		backupMgr, err := backup.NewManager(zapKV.Exec, backupDir, nodeID, 5*time.Minute, s3Cfg)
		if err != nil {
			logger.Warn("Failed to create backup manager", "err", err)
		} else {
			backupMgr.Start()
			defer backupMgr.Stop()
			logger.Info("Backup manager started", "period", "5m", "s3", s3Cfg != nil)
		}
	}

	// Start KMS-facing ZAP server. KMS dials this for threshold-signing
	// requests after running the ML-KEM-768 hybrid handshake. Empty
	// --kms-zap-listen disables it (e.g., for legacy compose stacks that
	// Default :9653 matches the the operator's zapPort so no
	// LiquidMPC CR change is needed. Trust at the network boundary
	// (NetworkPolicy + ZAP wire).
	kmsZapAddr := c.String("kms-zap-listen")
	if kmsZapAddr != "" {
		kmsZap, kmsZapErr := mpcapi.StartKMSZAP(mpcBackend, fmt.Sprintf("mpcd-kms-zap-%s", nodeID), kmsZapAddr)
		if kmsZapErr != nil {
			logger.Error("Failed to start KMS ZAP server", kmsZapErr, "addr", kmsZapAddr)
		} else {
			defer kmsZap.Stop()
			logger.Info("KMS ZAP server ready", "addr", kmsZapAddr, "nodeID", nodeID)
		}
	}

	// Start embedded threshold ZAP dispatcher
	// (luxfi/threshold/pkg/thresholdd). One process, one wire, seven
	// schemes (cggmp21/frost/pulsar/corona/magnetar/bls/doerner) —
	// consumed by teleport/mpc and any other client that speaks the
	// threshold bus.
	//
	// The dispatcher carries zero protocol policy on purpose: profile
	// gating and audit live on the API surface above. Auth and bind
	// scope are enforced HERE because the dispatcher would otherwise
	// be an anonymous local signing oracle (Red HIGH B1):
	//
	//   1. Bearer-token auth. Default token = SHA-256(privKey.Seed() ||
	//      "mpc-threshold-rpc"); deterministically derived from the
	//      node Ed25519 identity so every node in the cluster shares
	//      the same key without extra config. MPC_THRESHOLD_AUTH_TOKEN
	//      overrides for explicit operator control.
	//   2. Loopback-only bind. A non-loopback `threshold-listen`
	//      (operator typo `--threshold-listen 0.0.0.0:7301` or env
	//      override) is refused unless MPC_THRESHOLD_ALLOW_REMOTE=1
	//      is set, which forces operators to acknowledge the exposure.
	if thrAddr := c.String("threshold-listen"); thrAddr != "" {
		if !isThresholdLoopback(thrAddr) && os.Getenv("MPC_THRESHOLD_ALLOW_REMOTE") != "1" {
			logger.Fatal(
				"Threshold dispatcher refusing non-loopback bind without MPC_THRESHOLD_ALLOW_REMOTE=1",
				fmt.Errorf("addr=%s", thrAddr),
			)
		}

		// Derive / read the per-cluster bearer token.
		thrTok := os.Getenv("MPC_THRESHOLD_AUTH_TOKEN")
		if thrTok == "" {
			h := sha256.Sum256(append(privKey.Seed(), []byte("mpc-threshold-rpc")...))
			thrTok = hex.EncodeToString(h[:])
			logger.Warn("MPC_THRESHOLD_AUTH_TOKEN not set; derived threshold dispatcher token from node identity (set MPC_THRESHOLD_AUTH_TOKEN in production for cluster-wide consistency)")
		}

		// Split host:port — ZapServerConfig takes int port.
		_, thrPortStr, thrSplitErr := net.SplitHostPort(thrAddr)
		if thrSplitErr != nil {
			logger.Error("Failed to parse threshold-listen", thrSplitErr, "addr", thrAddr)
		} else {
			thrPort, thrAtoiErr := strconv.Atoi(thrPortStr)
			if thrAtoiErr != nil {
				logger.Error("Bad threshold-listen port", thrAtoiErr, "port", thrPortStr)
			} else {
				thrSrv, thrErr := thresholdd.NewZapServer(thresholdd.ZapServerConfig{
					NodeID:    "mpcd-threshold",
					Port:      thrPort,
					AuthToken: thrTok,
				})
				if thrErr != nil {
					logger.Error("Failed to build threshold dispatcher", thrErr)
				} else {
					// Strict-PQ gate (Red M-2): the dispatcher's
					// Sign_Ctx_Profile consults a ChainProfileResolver before
					// allowing the single-party dealer shortcut on ctx-bound
					// signing (pulsar/magnetar sign_ctx). With NO resolver
					// wired the gate fails OPEN — every chainID, strict-PQ
					// chains included, slips through. Wire the node's posture
					// BEFORE Start so the gate is live on the first accepted
					// sign_ctx, and fails CLOSED by default.
					profDef, profErr := chainProfileFromString(c.String("chain-profile"))
					if profErr != nil {
						logger.Fatal("Invalid --chain-profile", profErr)
					}
					thrSrv.SetChainProfileResolver(
						buildChainProfileResolver(profDef, splitCSV(c.String("legacy-chains"))),
					)

					if startErr := thrSrv.Start(); startErr != nil {
						logger.Error("Failed to start threshold dispatcher", startErr, "addr", thrAddr)
					} else {
						logger.Info(
							"Threshold dispatcher (ZAP) starting",
							"addr", thrAddr,
							"schemes", "cggmp21,frost,pulsar,corona,magnetar,doerner",
							"auth", "bearer-token (peer nodeID)",
							"strict_pq_default", profDef.String(),
							"legacy_chains", c.String("legacy-chains"),
						)
						defer thrSrv.Stop()
					}
				}
			}
		}
	}

	// Start Dashboard API server (ZapDB-backed, no external dependencies)
	apiListenAddr := c.String("api-listen")
	jwtSecret := c.String("jwt-secret")
	if jwtSecret == "" {
		jwtSecret = os.Getenv("MPC_JWT_SECRET")
	}
	if jwtSecret != "" {
		dbPath := filepath.Join(dataDir, "dashboard.db")
		database, err := db.New(dbPath, "")
		if err != nil {
			logger.Error("Failed to open dashboard database", err)
		} else {
			defer database.Close()

			apiServer := mpcapi.NewServer(database, mpcBackend, jwtSecret)
			apiServer.StartScheduler(ctx)

			// Wire HSM signer for intent co-signing.
			// Provider-specific config is folded into a single map; each
			// underlying provider reads only the keys it understands
			// and falls back to MPC_HSM_<PROVIDER>_* env vars for the
			// rest of its tuning surface.
			signerType := c.String("hsm-signer")
			if signerType != "" {
				signerConfig, cfgErr := buildHSMSignerConfig(c, signerType)
				if cfgErr != nil {
					logger.Error("Failed to build HSM signer config", cfgErr, "provider", signerType)
				} else {
					signer, signerErr := hsm.NewSigner(signerType, signerConfig)
					if signerErr != nil {
						logger.Error("Failed to create HSM signer", signerErr, "provider", signerType)
					} else {
						apiServer.SetHSM(signer)
						logger.Info("HSM signer configured for co-signing", "provider", signer.Provider())
					}
				}
			}

			// HSM threshold attestation (key share vault storage)
			if c.Bool("hsm-attest") {
				logger.Info("HSM threshold attestation enabled",
					"signer", c.String("hsm-signer"),
					"attest_key", c.String("hsm-signer-key-id"),
					"vault_provider", c.String("hsm-provider"),
				)
			}

			// Mount chi handler on Base
			os.Args = []string{"mpcd", "serve", "--http", apiListenAddr}
			baseApp := base.New()
			baseApp.OnServe().BindFunc(func(e *core.ServeEvent) error {
				// Embedded admin UI at /_/mpc/
				e.Router.GET("/_/mpc/{path...}", apis.Static(uimpc.DistDirFS(), true))

				e.Router.Any("/{path...}", func(re *core.RequestEvent) error {
					apiServer.Handler().ServeHTTP(re.Response, re.Request)
					return nil
				})
				return e.Next()
			})

			logger.Info("Dashboard API starting (Base+SQLite)", "addr", apiListenAddr)
			go func() {
				if err := baseApp.Start(); err != nil {
					logger.Error("Dashboard API server failed", err)
				}
			}()

			logger.Info("Dashboard API ready", "addr", apiListenAddr, "db", dbPath)
		}
	}

	// Wait for shutdown signal
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
	<-sigChan

	logger.Warn("Shutdown signal received, stopping...")
	return nil
}

// ConsensusMPCBackend implements api.MPCBackend using the consensus transport.
type ConsensusMPCBackend struct {
	pubSub       *ConsensusPubSubAdapter
	peerRegistry *ConsensusPeerRegistry
	factory      *transport.Factory
	keyInfoStore *transport.KeyInfoStore
	identity     *ConsensusIdentityStore
	nodeID       string
	threshold    int
}

func (b *ConsensusMPCBackend) TriggerKeygen(orgID, walletID string) (*mpcapi.KeygenResult, error) {
	if walletID == "" {
		h := sha256.Sum256([]byte(fmt.Sprintf("%s-%d", b.nodeID, time.Now().UnixNano())))
		walletID = hex.EncodeToString(h[:16])
	}

	resultTopic := fmt.Sprintf("mpc.mpc_keygen_result.%s", walletID)
	resultCh := make(chan *event.KeygenResultEvent, 1)
	unsub, err := b.pubSub.Subscribe(resultTopic, func(natMsg *nats.Msg) {
		var result event.KeygenResultEvent
		if err := json.Unmarshal(natMsg.Data, &result); err == nil {
			select {
			case resultCh <- &result:
			default:
			}
		}
	})
	if err != nil {
		return nil, fmt.Errorf("failed to subscribe to result topic: %w", err)
	}
	defer unsub.Unsubscribe()

	sig := b.identity.SignMessage([]byte(walletID))
	msg := types.GenerateKeyMessage{OrgID: orgID, WalletID: walletID, Signature: sig}
	msgData, _ := json.Marshal(msg)
	if err := b.pubSub.Publish("mpc:generate", msgData); err != nil {
		return nil, fmt.Errorf("failed to publish keygen: %w", err)
	}

	select {
	case result := <-resultCh:
		if result.ResultType != event.ResultTypeSuccess {
			return nil, fmt.Errorf("keygen failed: %s", result.ErrorReason)
		}
		evmAddr := ""
		btcAddr := ""
		solAddr := ""
		if len(result.ECDSAPubKey) >= 32 {
			evmAddr = pubKeyToEVMAddress(result.ECDSAPubKey)
			btcAddr = pubKeyToBtcAddress(result.ECDSAPubKey)
		}
		solAddr = eddsaPubKeyToSolAddress(result.EDDSAPubKey)
		// Report the security property this key actually has. Keygen accepts no
		// per-request threshold — the ring's --threshold governs — so a caller
		// can only verify what it asked for if we tell it what it got.
		return &mpcapi.KeygenResult{
			WalletID:     result.WalletID,
			ECDSAPubKey:  hex.EncodeToString(result.ECDSAPubKey),
			EDDSAPubKey:  hex.EncodeToString(result.EDDSAPubKey),
			EVMAddress:   evmAddr,
			BtcAddress:   btcAddr,
			SolAddress:   solAddr,
			Threshold:    b.threshold,
			Participants: b.peerRegistry.ConfiguredParticipants(),
		}, nil
	case <-time.After(120 * time.Second):
		return nil, fmt.Errorf("keygen timed out after 120s")
	}
}

func (b *ConsensusMPCBackend) TriggerSign(orgID, walletID string, payload []byte) (*mpcapi.SignResult, error) {
	txID := fmt.Sprintf("sign-%d", time.Now().UnixNano())
	resultTopic := fmt.Sprintf("mpc.mpc_signing_result.%s", walletID)
	resultCh := make(chan json.RawMessage, 1)
	unsub, err := b.pubSub.Subscribe(resultTopic, func(natMsg *nats.Msg) {
		select {
		case resultCh <- natMsg.Data:
		default:
		}
	})
	if err != nil {
		return nil, fmt.Errorf("failed to subscribe to signing result: %w", err)
	}
	defer unsub.Unsubscribe()

	// Look up key type from key info store
	keyType := types.KeyTypeSecp256k1 // default for ECDSA
	if b.keyInfoStore != nil {
		if info, err := b.keyInfoStore.Get(walletID); err == nil && info.KeyType != "" {
			keyType = types.KeyType(info.KeyType)
		}
	}
	// Normalize legacy key type names
	switch keyType {
	case "ecdsa", "ECDSA":
		keyType = types.KeyTypeSecp256k1
	case "eddsa", "EDDSA":
		keyType = types.KeyTypeEd25519
	}

	msg := types.SignTxMessage{
		OrgID:    orgID,
		KeyType:  keyType,
		WalletID: walletID,
		TxID:     txID,
		Tx:       payload,
	}
	// Sign the message with the node's private key
	raw, _ := msg.Raw()
	msg.Signature = b.identity.SignMessage(raw)
	msgData, _ := json.Marshal(msg)
	if err := b.pubSub.Publish("mpc:sign", msgData); err != nil {
		return nil, fmt.Errorf("failed to publish sign request: %w", err)
	}

	select {
	case data := <-resultCh:
		var result struct {
			ResultType        string `json:"result_type"`
			ErrorReason       string `json:"error_reason"`
			R                 []byte `json:"r"`
			S                 []byte `json:"s"`
			SignatureRecovery []byte `json:"signature_recovery"`
			Signature         []byte `json:"signature"`
		}
		if err := json.Unmarshal(data, &result); err != nil {
			return nil, fmt.Errorf("failed to unmarshal signing result: %w", err)
		}
		if result.ResultType == "error" {
			return nil, fmt.Errorf("MPC signing failed: %s", result.ErrorReason)
		}
		sigR := hex.EncodeToString(result.R)
		sigS := hex.EncodeToString(result.S)
		var sigHex, v string
		if len(result.Signature) > 0 {
			sigHex = hex.EncodeToString(result.Signature)
		}
		// Propagate the recovery id the signing session computed (0/1), and
		// assemble the canonical 65-byte r‖s‖v when the scheme is recoverable
		// (secp256k1) and no full blob was supplied — so KMS /sign never emits
		// an empty signature and callers get an ecrecover-ready value.
		if len(result.SignatureRecovery) > 0 {
			v = strconv.Itoa(int(result.SignatureRecovery[0]))
			if sigHex == "" && len(result.R) == 32 && len(result.S) == 32 {
				canonical := make([]byte, 0, 65)
				canonical = append(canonical, result.R...)
				canonical = append(canonical, result.S...)
				canonical = append(canonical, result.SignatureRecovery[0])
				sigHex = hex.EncodeToString(canonical)
			}
		}
		return &mpcapi.SignResult{R: sigR, S: sigS, V: v, Signature: sigHex}, nil
	case <-time.After(60 * time.Second):
		return nil, fmt.Errorf("signing timed out after 60s")
	}
}

func (b *ConsensusMPCBackend) TriggerReshare(orgID, walletID string, newThreshold int, newParticipants []string) error {
	msg := map[string]interface{}{
		"org_id":           orgID,
		"wallet_id":        walletID,
		"new_threshold":    newThreshold,
		"new_participants": newParticipants,
	}
	msgData, _ := json.Marshal(msg)
	return b.pubSub.Publish("mpc:reshare", msgData)
}

func (b *ConsensusMPCBackend) ExportKeyShare(orgID, walletID string) ([]byte, error) {
	key := mpc.OrgScopedKey(orgID, walletID)
	return b.factory.KVStore().Get(key)
}

func (b *ConsensusMPCBackend) GetClusterStatus() *mpcapi.ClusterStatus {
	ready := b.peerRegistry.ArePeersReady()
	connected := b.factory.Transport().GetPeers()
	return &mpcapi.ClusterStatus{
		NodeID:         b.nodeID,
		Mode:           "consensus",
		ExpectedPeers:  len(b.peerRegistry.peerIDs),
		ConnectedPeers: len(connected),
		Ready:          ready,
		Threshold:      b.threshold,
		Version:        Version,
	}
}

// loadOrGenerateIdentity loads or generates Ed25519 identity
func loadOrGenerateIdentity(keysDir, nodeID string) (ed25519.PrivateKey, ed25519.PublicKey, error) {
	identityPath := filepath.Join(keysDir, nodeID+"_identity.json")

	// Try to load existing identity
	data, err := os.ReadFile(identityPath)
	if err == nil {
		var identityData struct {
			NodeID     string `json:"node_id"`
			PublicKey  string `json:"public_key"`
			PrivateKey string `json:"private_key"`
		}
		if err := json.Unmarshal(data, &identityData); err == nil {
			privKeyBytes, err := hex.DecodeString(identityData.PrivateKey)
			if err == nil && len(privKeyBytes) == ed25519.PrivateKeySize {
				privKey := ed25519.PrivateKey(privKeyBytes)
				pubKey := privKey.Public().(ed25519.PublicKey)
				logger.Info("Loaded existing identity", "nodeID", nodeID)
				return privKey, pubKey, nil
			}
		}
	}

	// Generate new identity
	pubKey, privKey, err := ed25519.GenerateKey(nil)
	if err != nil {
		return nil, nil, err
	}

	// Save identity
	identityData := map[string]string{
		"node_id":     nodeID,
		"public_key":  hex.EncodeToString(pubKey),
		"private_key": hex.EncodeToString(privKey),
	}
	data, err = json.MarshalIndent(identityData, "", "  ")
	if err != nil {
		return nil, nil, err
	}
	if err := os.WriteFile(identityPath, data, 0600); err != nil {
		return nil, nil, err
	}

	logger.Info("Generated new identity", "nodeID", nodeID)
	return privKey, pubKey, nil
}

// ConsensusIdentityStore implements identity.Store for consensus mode
type ConsensusIdentityStore struct {
	nodeID          string
	privateKey      ed25519.PrivateKey
	publicKey       ed25519.PublicKey
	initiatorPubKey ed25519.PublicKey
	publicKeys      map[string][]byte
	mu              sync.RWMutex
}

func NewConsensusIdentityStore(nodeID string, privKey ed25519.PrivateKey, pubKey ed25519.PublicKey) *ConsensusIdentityStore {
	s := &ConsensusIdentityStore{
		nodeID:     nodeID,
		privateKey: privKey,
		publicKey:  pubKey,
		publicKeys: make(map[string][]byte),
	}
	s.publicKeys[nodeID] = pubKey

	// Load the event initiator public key from viper config.
	// This Ed25519 public key is used to verify that inbound event
	// messages (keygen, signing, reshare) originated from the authorized
	// initiator and have not been tampered with.
	if initiatorHex := viper.GetString("event_initiator_pubkey"); initiatorHex != "" {
		if decoded, err := hex.DecodeString(initiatorHex); err == nil && len(decoded) == ed25519.PublicKeySize {
			s.initiatorPubKey = ed25519.PublicKey(decoded)
		}
	}

	// In consensus mode, if no explicit initiator key is configured,
	// skip initiator verification since all messages come from within
	// the trusted cluster. The internal API (port 9800) is not
	// exposed externally.
	if s.initiatorPubKey == nil {
		logger.Info("No explicit initiator key configured; initiator verification will be skipped (consensus mode)")
	}

	return s
}

// SignMessage signs a message payload with the node's private key.
// Used by HTTP endpoints to sign event messages before publishing.
func (s *ConsensusIdentityStore) SignMessage(payload []byte) []byte {
	return ed25519.Sign(s.privateKey, payload)
}

func (s *ConsensusIdentityStore) GetPublicKey(nodeID string) ([]byte, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if key, ok := s.publicKeys[nodeID]; ok {
		return key, nil
	}
	return nil, fmt.Errorf("public key not found for node: %s", nodeID)
}

func (s *ConsensusIdentityStore) VerifyInitiatorMessage(msg types.InitiatorMessage) error {
	// In consensus mode without an explicit initiator key, skip
	// verification. The internal API is only accessible within the
	// cluster, so all messages are trusted.
	if s.initiatorPubKey == nil {
		return nil
	}

	// Reconstruct the canonical payload that was signed (excludes the
	// signature field itself).
	raw, err := msg.Raw()
	if err != nil {
		return fmt.Errorf("failed to get raw message data: %w", err)
	}

	sig := msg.Sig()
	if len(sig) == 0 {
		return fmt.Errorf("message has no signature")
	}

	if !ed25519.Verify(s.initiatorPubKey, raw, sig) {
		return fmt.Errorf("invalid Ed25519 signature from initiator")
	}

	return nil
}

func (s *ConsensusIdentityStore) AddPeerPublicKey(nodeID string, pubKey []byte) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.publicKeys[nodeID] = pubKey
}

// SignWireMessage signs a protocol wire message with this node's Ed25519 key.
func (s *ConsensusIdentityStore) SignWireMessage(msg *types.Message) {
	msg.Sign(s.privateKey)
}

// VerifyWireMessage verifies a protocol wire message's Ed25519 signature
// using the sender's public key looked up by SenderNodeID.
func (s *ConsensusIdentityStore) VerifyWireMessage(msg *types.Message) error {
	if len(msg.Signature) == 0 {
		return fmt.Errorf("message has no signature")
	}
	nodeID := msg.SenderNodeID
	if nodeID == "" {
		nodeID = msg.SenderID
	}
	pubKey, err := s.GetPublicKey(nodeID)
	if err != nil {
		return fmt.Errorf("failed to get sender's public key for node %s: %w", nodeID, err)
	}
	return msg.Verify(ed25519.PublicKey(pubKey))
}

// ConsensusPubSubAdapter adapts transport.PubSub to messaging.PubSub
type ConsensusPubSubAdapter struct {
	pubsub *transport.PubSub
}

func NewConsensusPubSubAdapter(pubsub *transport.PubSub) *ConsensusPubSubAdapter {
	return &ConsensusPubSubAdapter{pubsub: pubsub}
}

func (a *ConsensusPubSubAdapter) Publish(topic string, data []byte) error {
	return a.pubsub.Publish(topic, data)
}

func (a *ConsensusPubSubAdapter) PublishWithReply(topic, reply string, data []byte, headers map[string]string) error {
	return a.pubsub.PublishWithReply(topic, reply, data, headers)
}

func (a *ConsensusPubSubAdapter) Subscribe(topic string, handler func(msg *nats.Msg)) (messaging.Subscription, error) {
	sub, err := a.pubsub.Subscribe(topic, func(msg *transport.NATSMsg) {
		// Convert transport.NATSMsg to nats.Msg
		natsMsg := &nats.Msg{
			Subject: msg.Subject,
			Reply:   msg.Reply,
			Data:    msg.Data,
			Header:  nats.Header(msg.Header),
		}
		handler(natsMsg)
	})
	if err != nil {
		return nil, err
	}
	return &consensusSubscription{sub: sub}, nil
}

type consensusSubscription struct {
	sub *transport.Subscription
}

func (s *consensusSubscription) Unsubscribe() error {
	return s.sub.Unsubscribe()
}

// ConsensusPeerRegistry adapts transport.Registry to mpc.PeerRegistry
type ConsensusPeerRegistry struct {
	registry *transport.Registry
	nodeID   string
	peerIDs  []string
}

func NewConsensusPeerRegistry(registry *transport.Registry, nodeID string, peerIDs []string) *ConsensusPeerRegistry {
	return &ConsensusPeerRegistry{
		registry: registry,
		nodeID:   nodeID,
		peerIDs:  peerIDs,
	}
}

func (r *ConsensusPeerRegistry) Ready() error {
	return r.registry.Ready()
}

func (r *ConsensusPeerRegistry) Resign() error {
	return r.registry.Resign()
}

func (r *ConsensusPeerRegistry) WatchPeersReady() {
	r.registry.WatchPeersReady()
}

func (r *ConsensusPeerRegistry) ArePeersReady() bool {
	return r.registry.ArePeersReady()
}

// HasSigningQuorum delegates to the underlying transport registry. See
// pkg/transport/registry.go for the rationale (ready peer count >= threshold+1).
func (r *ConsensusPeerRegistry) HasSigningQuorum(threshold int) bool {
	return r.registry.HasSigningQuorum(threshold)
}

func (r *ConsensusPeerRegistry) GetReadyPeersCount() int64 {
	return r.registry.GetReadyPeersCount()
}

func (r *ConsensusPeerRegistry) GetTotalPeersCount() int64 {
	return int64(len(r.peerIDs) + 1) // peers + self
}

func (r *ConsensusPeerRegistry) GetReadyPeersIncludeSelf() []string {
	return r.registry.GetReadyPeersIncludeSelf()
}

// ConfiguredParticipants returns the party set this node is configured for
// (peers + self), sorted so every node reports the same list for the same key.
//
// Deliberately NOT GetReadyPeersIncludeSelf: readiness fluctuates, so it answers
// "who is up right now", whereas a key's participant set is a property of the
// key itself and must not change because a pod restarted.
func (r *ConsensusPeerRegistry) ConfiguredParticipants() []string {
	out := make([]string, 0, len(r.peerIDs)+1)
	out = append(out, r.peerIDs...)
	out = append(out, r.nodeID)
	sort.Strings(out)
	return out
}

// ConsensusKeyInfoStore adapts transport.KeyInfoStore to keyinfo.Store
type ConsensusKeyInfoStore struct {
	store        *transport.KeyInfoStore
	peerRegistry *ConsensusPeerRegistry
}

func NewConsensusKeyInfoStore(store *transport.KeyInfoStore, peerRegistry *ConsensusPeerRegistry) *ConsensusKeyInfoStore {
	return &ConsensusKeyInfoStore{store: store, peerRegistry: peerRegistry}
}

func (s *ConsensusKeyInfoStore) Get(walletID string) (*keyinfo.KeyInfo, error) {
	info, err := s.store.Get(walletID)
	if err != nil {
		return nil, err
	}
	// Convert transport.KeyInfo to keyinfo.KeyInfo
	// Populate ParticipantPeerIDs from peer registry (all ready peers including self)
	participantPeerIDs := s.peerRegistry.GetReadyPeersIncludeSelf()
	return &keyinfo.KeyInfo{
		ParticipantPeerIDs: participantPeerIDs,
		Threshold:          info.Threshold,
		Version:            1, // Default version
	}, nil
}

func (s *ConsensusKeyInfoStore) Save(walletID string, info *keyinfo.KeyInfo) error {
	return s.store.RegisterKey(walletID, "secp256k1", info.Threshold, "", "", nil)
}

// ConsensusMessageQueue adapts transport for messaging.MessageQueue
type ConsensusMessageQueue struct {
	transport *transport.Transport
	nodeID    string
	queueType string
	handlers  map[string]func([]byte) error
	mu        sync.RWMutex
}

func NewConsensusMessageQueue(t *transport.Transport, nodeID, queueType string) *ConsensusMessageQueue {
	return &ConsensusMessageQueue{
		transport: t,
		nodeID:    nodeID,
		queueType: queueType,
		handlers:  make(map[string]func([]byte) error),
	}
}

func (q *ConsensusMessageQueue) Enqueue(topic string, message []byte, options *messaging.EnqueueOptions) error {
	// Broadcast the message via transport's Publish method
	return q.transport.Publish(topic, message)
}

func (q *ConsensusMessageQueue) Dequeue(topic string, handler func(message []byte) error) error {
	q.mu.Lock()
	q.handlers[topic] = handler
	q.mu.Unlock()
	// In consensus mode, messages are delivered via PubSub subscriptions
	// The handler will be called when messages arrive
	return nil
}

func (q *ConsensusMessageQueue) Close() {
	// Nothing to close in consensus mode
}

// pubKeyToEVMAddress derives an Ethereum address from an ECDSA public key.
// Accepts compressed (33 bytes), uncompressed (65 bytes), or raw x-coordinate (32 bytes).
func pubKeyToEVMAddress(pubKey []byte) string {
	var xyBytes []byte // 64 bytes: X(32) || Y(32)
	switch len(pubKey) {
	case 65:
		// Uncompressed: 0x04 || X(32) || Y(32)
		xyBytes = pubKey[1:]
	case 33:
		// Compressed: 0x02/0x03 || X(32) — decompress via secp256k1.
		x, y := ellipticUnmarshalCompressed(pubKey)
		if x == nil {
			return ""
		}
		// LEFT-PAD each coordinate to 32 bytes. big.Int.Bytes() drops leading
		// zero bytes, so a coordinate < 2^248 (X or Y with a high zero byte,
		// ~1/256 of keys) would make xyBytes != 64 and the len guard below return
		// "" — an EMPTY address for an otherwise valid key. FillBytes zero-pads
		// into a fixed 32-byte slice. (case 32 already pads; case 33 did not.)
		xb := make([]byte, 32)
		yb := make([]byte, 32)
		x.FillBytes(xb)
		y.FillBytes(yb)
		xyBytes = append(xb, yb...)
	case 32:
		// Raw x-coordinate only — try decompressing with 0x02 prefix (even y)
		compressed := append([]byte{0x02}, pubKey...)
		x, y := ellipticUnmarshalCompressed(compressed)
		if x == nil {
			// Try odd y
			compressed[0] = 0x03
			x, y = ellipticUnmarshalCompressed(compressed)
		}
		if x == nil {
			return ""
		}
		xBytes := make([]byte, 32)
		yBytes := make([]byte, 32)
		xB := x.Bytes()
		yB := y.Bytes()
		copy(xBytes[32-len(xB):], xB)
		copy(yBytes[32-len(yB):], yB)
		xyBytes = append(xBytes, yBytes...)
	default:
		// Try as hex string
		decoded, err := hex.DecodeString(string(pubKey))
		if err == nil && len(decoded) > 0 {
			return pubKeyToEVMAddress(decoded)
		}
		return ""
	}
	if len(xyBytes) != 64 {
		return ""
	}
	hash := sha3.NewLegacyKeccak256()
	hash.Write(xyBytes)
	addrBytes := hash.Sum(nil)[12:]
	return "0x" + hex.EncodeToString(addrBytes)
}

// ellipticUnmarshalCompressed decompresses a secp256k1 compressed public key.
func ellipticUnmarshalCompressed(compressed []byte) (*big.Int, *big.Int) {
	if len(compressed) != 33 || (compressed[0] != 0x02 && compressed[0] != 0x03) {
		return nil, nil
	}
	curve := crypto_elliptic.P256() // Use P256 as base; secp256k1 params below
	// secp256k1 curve parameters
	p, _ := new(big.Int).SetString("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F", 16)
	x := new(big.Int).SetBytes(compressed[1:33])
	// y² = x³ + 7 (mod p) for secp256k1
	x3 := new(big.Int).Mul(x, x)
	x3.Mul(x3, x)
	x3.Mod(x3, p)
	y2 := new(big.Int).Add(x3, big.NewInt(7))
	y2.Mod(y2, p)
	// ModSqrt
	y := new(big.Int).ModSqrt(y2, p)
	if y == nil {
		return nil, nil
	}
	// Check parity
	if y.Bit(0) != uint(compressed[0]&1) {
		y.Sub(p, y)
	}
	_ = curve // suppress unused
	return x, y
}

// pubKeyToBtcAddress derives a Bitcoin P2PKH address from a secp256k1 public key.
// Accepts compressed (33 bytes), uncompressed (65 bytes), or raw x-coordinate (32 bytes).
func pubKeyToBtcAddress(pubKey []byte) string {
	var compressed []byte
	switch len(pubKey) {
	case 33:
		compressed = pubKey
	case 65:
		// Compress: take X coordinate, prefix with 0x02 or 0x03 based on Y parity
		prefix := byte(0x02)
		if pubKey[64]&1 == 1 {
			prefix = 0x03
		}
		compressed = append([]byte{prefix}, pubKey[1:33]...)
	case 32:
		// Raw x-coordinate — use even y (0x02)
		compressed = append([]byte{0x02}, pubKey...)
	default:
		return ""
	}
	// SHA256(compressed pubkey)
	sha := sha256.Sum256(compressed)
	// RIPEMD160(SHA256)
	rip := ripemd160.New()
	rip.Write(sha[:])
	pubKeyHash := rip.Sum(nil) // 20 bytes

	// Base58Check encode with version byte 0x00 (mainnet P2PKH)
	return base58CheckEncode(0x00, pubKeyHash)
}

// base58CheckEncode encodes data with a version byte using Base58Check encoding.
func base58CheckEncode(version byte, payload []byte) string {
	versioned := append([]byte{version}, payload...)
	// Double SHA256 checksum
	first := sha256.Sum256(versioned)
	second := sha256.Sum256(first[:])
	checksum := second[:4]
	full := append(versioned, checksum...)
	return base58.Encode(full)
}

// eddsaPubKeyToSolAddress derives a Solana address from an Ed25519 public key,
// or returns "" when the bytes are not one.
//
// The check is delegated to pkg/address, which validates the curve rather than
// the length. Length alone cannot do this job: a BIP-340 x-only secp256k1 key is
// also 32 bytes and base58-encodes into an address that accepts deposits and can
// never be spent from. "" means the caller must omit sol_address entirely.
func eddsaPubKeyToSolAddress(pubKey []byte) string {
	if len(pubKey) == 0 {
		// No Ed25519 leg for this wallet. Ordinary and expected, so it is not
		// worth a warning on every response.
		return ""
	}
	addr, err := address.Solana(pubKey)
	if err != nil {
		// A key that is present but not an Ed25519 key is a different matter:
		// something upstream produced the wrong curve, and that is worth
		// shouting about even though it has been refused here.
		logger.Warn("Refusing to derive a Solana address from a non-Ed25519 key",
			"error", err.Error(),
			"consequence", "sol_address omitted from the response",
		)
		return ""
	}
	return addr
}

// isThresholdLoopback reports whether `--threshold-listen` resolves to
// a loopback host. Bare `:port` (= 0.0.0.0:port) is NOT loopback —
// operators must spell out the bind host. See the dispatcher startup
// block above for the rationale (Red HIGH B1).
func isThresholdLoopback(addr string) bool {
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		return false
	}
	if host == "" {
		return false
	}
	if host == "localhost" {
		return true
	}
	if strings.EqualFold(host, "::1") {
		return true
	}
	ip := net.ParseIP(host)
	if ip == nil {
		return false
	}
	return ip.IsLoopback()
}