// Package runtimeprocess owns lifecycle composition for independently runnable
// NexusCRM Runtime roles. It contains no engine, product, or work scheduler.
package runtimeprocess

import (
	"context"
	"errors"
	"net"
	"net/http"
	"net/http/pprof"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	"github.com/prometheus/client_golang/prometheus"
)

type State int32

const (
	StateAssembled State = iota
	StateStarted
	StateReady
	StateDraining
	StateStopped
)

type Config struct {
	Role         string
	Version      string
	AdminAddress string
	GRPCAddress  string
	GRPCServer   RPCServer
	Capabilities []string
	Ready        func() bool
	BeginDrain   func()
	Admission    func() AdmissionSnapshot
	// MetricsCollector is an optional role-owned, bounded-label collector
	// registered in the process's private registry. Lifecycle metrics remain
	// owned here; engine/adapter metrics remain owned by their role.
	MetricsCollector prometheus.Collector
	// AdditionalMetricsCollectors are further role-owned collectors registered in the same
	// private registry.
	//
	// It exists because the registry is private: a metric registered on the default registry
	// is simply invisible in this process, so anything an operator has to see here must be
	// handed over explicitly. One slot was enough while the records server was the only owner;
	// a second owner would otherwise have to displace it, and a metric silently absent from a
	// process is worse than one that was never written.
	AdditionalMetricsCollectors []prometheus.Collector
	// Controller is process-owned background reconciliation. It starts only
	// after every listener is established and is canceled and joined by Stop.
	Controller func(context.Context) error
	// ExposeDiagnostics serves Go's on-demand profiles only on the private,
	// loopback admin listener. Merely registering these handlers does not enable
	// block/mutex sampling or verbose request timing.
	ExposeDiagnostics bool
	// EnableProfiling additionally enables continuous block/mutex sampling for
	// an explicitly profiled process. It retains the historical implication that
	// the diagnostic handlers are exposed, but is intentionally not required for
	// bounded CPU/heap/goroutine capture.
	EnableProfiling bool
}

type AdmissionSnapshot struct {
	Limit            int
	InFlight         int
	Accepted         uint64
	RejectedFull     uint64
	RejectedDraining uint64
}

// readinessObservation is the latest result produced by an active readiness
// probe. Observability callers consume this immutable snapshot instead of
// repeating role-owned dependency checks.
type readinessObservation struct {
	Ready     bool
	Known     bool
	CheckedAt time.Time
}

// RPCServer is the lifecycle surface needed by the API role. Keeping this
// interface transport-library-free prevents the no-work worker binary from
// acquiring a gRPC dependency merely because the roles share lifecycle code.
type RPCServer interface {
	Serve(net.Listener) error
	GracefulStop()
	Stop()
}

// Process is an explicit assemble/start/ready/drain/stop/wait state machine.
type Process struct {
	config Config
	state  atomic.Int32

	mu               sync.Mutex
	adminServer      *http.Server
	adminHandler     http.Handler
	adminListener    net.Listener
	grpcListener     net.Listener
	done             chan error
	stopOnce         sync.Once
	controllerCancel context.CancelFunc
	controllerDone   chan struct{}
	controllerAlive  atomic.Bool

	readinessSequence atomic.Uint64
	readinessMu       sync.RWMutex
	readinessObserved uint64
	readiness         readinessObservation
}

func Assemble(config Config) (*Process, error) {
	if config.Role == "" || config.Version == "" || config.AdminAddress == "" {
		return nil, errors.New("runtime process: role, version, and admin address are required")
	}
	if (config.GRPCServer == nil) != (config.GRPCAddress == "") {
		return nil, errors.New("runtime process: gRPC server and address must be configured together")
	}
	if config.GRPCServer != nil && config.BeginDrain == nil {
		return nil, errors.New("runtime process: gRPC roles require a begin-drain admission hook")
	}
	if (config.ExposeDiagnostics || config.EnableProfiling) && !IsLoopbackListenAddress(config.AdminAddress) {
		return nil, errors.New("runtime process: profiling requires a loopback admin address")
	}
	p := &Process{config: config, done: make(chan error, 3)}
	operationalHandler, err := processhttp.NewHandler(processhttp.Config{
		Role: config.Role, Version: config.Version,
		Live: p.Live, Ready: p.Ready, Capabilities: config.Capabilities,
	})
	if err != nil {
		return nil, err
	}
	metricsHandler, err := newMetricsHandler(p)
	if err != nil {
		return nil, err
	}
	adminMux := http.NewServeMux()
	adminMux.Handle(MetricsPath, metricsHandler)
	if config.EnableProfiling {
		// A one-millisecond block sampling rate retains the long waits relevant to
		// Test2 without recording every short scheduler handoff. Mutex fraction 1
		// is scoped to the explicitly profiled process lifetime.
		runtime.SetBlockProfileRate(1_000_000)
		runtime.SetMutexProfileFraction(1)
	}
	if config.ExposeDiagnostics || config.EnableProfiling {
		adminMux.HandleFunc("/debug/pprof/", pprof.Index)
		adminMux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
		adminMux.HandleFunc("/debug/pprof/profile", pprof.Profile)
		adminMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
		adminMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
		for _, profile := range []string{"allocs", "block", "goroutine", "heap", "mutex", "threadcreate"} {
			adminMux.Handle("/debug/pprof/"+profile, pprof.Handler(profile))
		}
	}
	adminMux.Handle("/", operationalHandler)
	p.adminHandler = adminMux
	p.state.Store(int32(StateAssembled))
	return p, nil
}

// IsLoopbackListenAddress reports whether an HTTP listener is confined to the
// local host. Callers use the same predicate as Assemble when deciding whether
// low-overhead diagnostic handlers may be exposed on an operational listener.
func IsLoopbackListenAddress(address string) bool {
	host, _, err := net.SplitHostPort(address)
	if err != nil {
		return false
	}
	if host == "localhost" {
		return true
	}
	ip := net.ParseIP(host)
	return ip != nil && ip.IsLoopback()
}

func (p *Process) State() State { return State(p.state.Load()) }

func (p *Process) Live() bool {
	state := p.State()
	return state >= StateStarted && state < StateStopped
}

func (p *Process) Ready() bool {
	sequence := p.readinessSequence.Add(1)
	ready := p.localReady()
	if ready && p.config.Ready != nil {
		ready = p.config.Ready()
	}
	p.recordReadiness(sequence, ready, time.Now())
	return ready
}

func (p *Process) localReady() bool {
	return p.State() == StateReady && (p.config.Controller == nil || p.controllerAlive.Load())
}

func (p *Process) recordReadiness(sequence uint64, ready bool, checkedAt time.Time) {
	p.readinessMu.Lock()
	defer p.readinessMu.Unlock()
	if sequence < p.readinessObserved {
		return
	}
	p.readinessObserved = sequence
	p.readiness = readinessObservation{Ready: ready, Known: true, CheckedAt: checkedAt}
}

func (p *Process) resetReadinessObservation() {
	p.readinessMu.Lock()
	p.readinessObserved = p.readinessSequence.Load()
	p.readiness = readinessObservation{}
	p.readinessMu.Unlock()
}

// readinessSnapshot is memory-only. It deliberately never calls Config.Ready:
// /readyz owns active dependency validation, while /metrics only reports the
// latest completed validation and the current local lifecycle fence.
func (p *Process) readinessSnapshot() readinessObservation {
	if !p.localReady() {
		return readinessObservation{Known: true}
	}
	if p.config.Ready == nil {
		return readinessObservation{Ready: true, Known: true}
	}
	p.readinessMu.RLock()
	defer p.readinessMu.RUnlock()
	return p.readiness
}

func (p *Process) Start() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.State() != StateAssembled {
		return errors.New("runtime process: start requires assembled state")
	}
	adminListener, err := net.Listen("tcp", p.config.AdminAddress)
	if err != nil {
		return err
	}
	p.adminListener = adminListener
	if p.config.GRPCServer != nil {
		p.grpcListener, err = net.Listen("tcp", p.config.GRPCAddress)
		if err != nil {
			_ = adminListener.Close()
			p.adminListener = nil
			return err
		}
	}
	p.adminServer = &http.Server{Handler: p.adminHandler}
	p.state.Store(int32(StateStarted))
	go func() {
		err := p.adminServer.Serve(adminListener)
		if errors.Is(err, http.ErrServerClosed) {
			err = nil
		}
		p.done <- err
	}()
	if p.config.GRPCServer != nil {
		go func() {
			err := p.config.GRPCServer.Serve(p.grpcListener)
			if p.State() >= StateDraining {
				err = nil
			}
			p.done <- err
		}()
	}
	if p.config.Controller != nil {
		controllerCtx, cancel := context.WithCancel(context.Background())
		p.controllerCancel = cancel
		p.controllerDone = make(chan struct{})
		p.controllerAlive.Store(true)
		go func() {
			defer close(p.controllerDone)
			err := p.config.Controller(controllerCtx)
			p.controllerAlive.Store(false)
			if controllerCtx.Err() == nil {
				p.done <- err
			}
		}()
	}
	p.state.Store(int32(StateReady))
	// Checks made while the process was assembled or starting did not validate
	// dependencies in the ready lifecycle and must not seed the ready-state
	// metric observation.
	p.resetReadinessObservation()
	return nil
}

func (p *Process) BeginDrain() {
	for {
		state := p.State()
		if state >= StateDraining {
			return
		}
		if p.state.CompareAndSwap(int32(state), int32(StateDraining)) {
			if p.config.BeginDrain != nil {
				p.config.BeginDrain()
			}
			return
		}
	}
}

func (p *Process) Stop(ctx context.Context) error {
	var stopErr error
	p.stopOnce.Do(func() {
		p.BeginDrain()
		if p.controllerCancel != nil {
			p.controllerCancel()
		}
		if p.config.GRPCServer != nil {
			stopped := make(chan struct{})
			go func() {
				p.config.GRPCServer.GracefulStop()
				close(stopped)
			}()
			select {
			case <-stopped:
			case <-ctx.Done():
				p.config.GRPCServer.Stop()
				<-stopped
				stopErr = ctx.Err()
			}
		}
		if p.adminServer != nil {
			if err := p.adminServer.Shutdown(ctx); err != nil && stopErr == nil {
				stopErr = err
			}
		}
		if p.controllerDone != nil {
			select {
			case <-p.controllerDone:
			case <-ctx.Done():
				if stopErr == nil {
					stopErr = ctx.Err()
				}
			}
		}
		p.state.Store(int32(StateStopped))
	})
	return stopErr
}

// Wait returns the first unexpected serving result. Callers normally select it
// against their signal context and then invoke Stop.
func (p *Process) Wait() error { return <-p.done }

func (p *Process) AdminAddress() string {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.adminListener == nil {
		return ""
	}
	return p.adminListener.Addr().String()
}

func (p *Process) GRPCAddress() string {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.grpcListener == nil {
		return ""
	}
	return p.grpcListener.Addr().String()
}