package activator

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/registry"
)

// CmdRunDirName is the directory under an instance's own data directory
// that holds each run's --file uploads (run spec §5.2 step 3).
const CmdRunDirName = ".carlos-run"

// cmdErrSocket is the stderr listener's name; stdout's is the role's.
const cmdErrSocket = "cmd-err"

// cmdDrainBound is how long a stream may take to reach EOF once its unit
// has stopped.
const cmdDrainBound = 5 * time.Second

// ErrCmdNotNested: the instance is not on the nested layout, so a command
// would run with the agent's reach (run spec §5.2 step 1, refused "flat").
var ErrCmdNotNested = errors.New("activator: the instance is not on the nested layout")

// CmdSpec is one run's launch, as the box's run pass hands it over.
type CmdSpec struct {
	RunID string
	// Args is exactly what followed `--`; the binary is the host's live link.
	Args []string
	// RunDir is CARLOS_RUN_DIR: <instance dir>/.carlos-run/<RunID>, made and
	// filled by the caller.
	RunDir string
	// Stdout and Stderr receive the unit's two streams, byte for byte.
	Stdout, Stderr io.Writer
	// Prepare runs after the run is attached to the awake entry and before
	// the unit starts (Service.RunCmd); an error ends the run before launch.
	Prepare func(ctx context.Context) error
	// Recheck, when set, is the run's launch eligibility checked again
	// after RunCmd has waited for a swap's switch (re-review 5): Prepare's
	// last check is that old by then. An error ends the run before launch,
	// as Prepare's does.
	Recheck func(ctx context.Context) error

	// serving is the serving generation's staged build, set by RunCmd at its
	// last awake check when that generation names one, generations on or
	// off (zero-downtime spec §7.5 and settings ruling R-EXEC-SIDECAR: what
	// runs beside an instance is its serving artifact). Nil runs the live
	// link. Callers cannot set it.
	serving *instancectl.StagedExe
	// genEnv is the file environment the run's generation launched with,
	// from its handle (unitProc.LaunchEnv), set by RunCmd with serving
	// (merge ruling M3). The run's variables are composed over it, never
	// over the delivered files, which may hold a config revision that
	// generation never activated. StartCmd refuses a run without one.
	// Callers cannot set it.
	genEnv map[string]string
	// genKeyLayer is the layer genEnv's CARLOS_INSTANCE_KEY came from, from
	// the same handle (unitProc.LaunchKeyLayer), for the replica gate.
	// Callers cannot set it.
	genKeyLayer instancekey.Layer
}

// errNoGenEnv: the run's generation kept no launch environment, so the run
// would have to read the delivered files, which may differ from what the
// instance is serving with. It is refused instead (merge ruling M3).
var errNoGenEnv = errors.New("activator: the serving generation kept no launch environment for a run")

// cmdExeStarter is the cmd start with the executable named (Client.CmdExe).
// A separate interface, like GenControl, so a Control without it still
// satisfies Control; StartCmd asks for it only when a run names a build.
type cmdExeStarter interface {
	CmdExe(ctx context.Context, host string, memoryMax int64, x instancectl.StagedExe) (string, error)
}

var _ cmdExeStarter = (*instancectl.Client)(nil)

// errNoCmdExe: a run names its serving generation's build and the control
// channel cannot carry it. The run is refused rather than run on the live
// link, which after a fallback names a build the instance is not serving.
var errNoCmdExe = errors.New("activator: the instance-control client cannot name a run's build")

// CmdUnit is a started cmd unit.
type CmdUnit interface {
	Invocation() string
	// Poll long-polls the invocation, at most about 30 seconds.
	Poll(ctx context.Context) (instancectl.CmdStatus, error)
	// StopAndSettle stops the unit (SIGTERM, then SIGKILL after 20 s),
	// proves its cgroup empty and reads both streams to their end. It is
	// safe to call more than once: once a call has succeeded, later ones
	// return nil at once, and a call after a failure tries again (merge
	// ruling M1).
	StopAndSettle(ctx context.Context) error
}

// CmdBackend is the backend capability Service.RunCmd needs. StartCmd may
// return a unit together with an error: the start failed after the unit
// may have begun, and the unit must still be stopped and settled.
type CmdBackend interface {
	StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error)
}

var _ CmdBackend = (*ExecBackend)(nil)

// cmdVars is a run's environment over fileEnv, the file environment its
// generation launched with (run spec §4, codex 7; merge ruling M3). That
// environment carries CARLOS_INSTANCE_KEY and CARLOS_ADMIN_TOKEN through:
// the command runs as the instance, on the instance's sealed data, with
// the key the serving generation holds and never one of its own. Over it:
// the control-channel variables stripped as seedVars strips them,
// CARLOS_REGION set or removed from the backend's region as agentOwnedVars
// does, CARLOS_HOST from the row, and the three run variables.
func (b *ExecBackend) cmdVars(rt registry.Route, fileEnv map[string]string, dbPath, runID, runDir string) map[string]string {
	v := withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE")
	if b.Region != "" {
		v = withVar(v, "CARLOS_REGION", b.Region)
	} else {
		v = withoutVars(v, "CARLOS_REGION")
	}
	v = withVar(v, "CARLOS_HOST", rt.Host)
	v = withVar(v, "CARLOS_DB", dbPath)
	v = withVar(v, "CARLOS_RUN_ID", runID)
	return withVar(v, "CARLOS_RUN_DIR", runDir)
}

// checkCmdRunDir refuses a run id that is not one plain path element and a
// run directory other than <instance dir>/.carlos-run/<RunID>. The run
// directory is deleted after the run, so it must never name anything
// outside the instance's own .carlos-run. It does no disk work.
func checkCmdRunDir(dataDir string, spec CmdSpec) error {
	id := spec.RunID
	if id == "" || id == "." || id == ".." || filepath.Base(id) != id || strings.ContainsAny(id, `/\`) {
		return fmt.Errorf("run id %q is not a single path element", id)
	}
	root := filepath.Join(dataDir, CmdRunDirName)
	if want := filepath.Join(root, id); spec.RunDir != want || filepath.Dir(spec.RunDir) != root {
		return fmt.Errorf("run directory %q is not %q", spec.RunDir, want)
	}
	return nil
}

func (b *ExecBackend) cmdLayout(rt registry.Route) (Layout, string) {
	if b.layoutFn != nil {
		return b.layoutFn(rt)
	}
	return b.Layout(rt)
}

// StartCmd starts one run's cmd unit (run spec §5.4): the layout check, the
// nested gate, the live link, the instance's directories, the environment
// its generation launched with (spec.genEnv, merge ruling M3), the two
// stream listeners, launch.cmd.json, then POST /cmd. It returns
// once the unit has an invocation. The environment is never logged. A
// failure after the POST was sent returns a unit with the error: the unit
// may be running, and the caller must stop and settle it.
//
// A run that names its serving generation's staged build (spec.serving)
// has root mount that build instead of the live link, and needs no live
// link of its own. The argv is the same /opt/carlos/live/<host> inside the
// namespace either way.
func (b *ExecBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
	if l, why := b.cmdLayout(rt); l != LayoutNested {
		return nil, fmt.Errorf("%w: %s is %s (%s)", ErrCmdNotNested, rt.Host, l, why)
	}
	if err := checkCmdRunDir(b.pathsFor(rt).dataDir, spec); err != nil {
		return nil, fmt.Errorf("%s: %w", rt.Host, err)
	}
	if err := b.nestedGate(rt); err != nil {
		return nil, err
	}
	var exeStart cmdExeStarter
	if spec.serving != nil {
		cs, ok := b.Control.(cmdExeStarter)
		if !ok {
			return nil, fmt.Errorf("%s: %w", rt.Host, errNoCmdExe)
		}
		exeStart = cs
	} else if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
		return nil, err
	}
	if spec.genEnv == nil {
		return nil, fmt.Errorf("%s: %w", rt.Host, errNoGenEnv)
	}
	// A run's command is a launch like any other (instance-key rulings 26
	// to 29): on a replica it starts only with its home's key, checked on
	// the environment it is about to be given and the layer that key came
	// from. A command has no control socket (cmdVars strips it), so a
	// replica's command with no key is refused, as a seed or a worker is:
	// an app on the instance-key helper would otherwise mint its own key
	// over the home's data. Nothing is laid out or started before this.
	if err := b.replicaKeyGate(rt, spec.genEnv, spec.genKeyLayer, false); err != nil {
		return nil, err
	}
	p, _, _, _, err := b.prepare(rt, instancectl.RoleCmd)
	if err != nil {
		return nil, err
	}
	out, err := listenCmdStream(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), spec.Stdout)
	if err != nil {
		return nil, fmt.Errorf("stdout listener for %s: %w", rt.Host, err)
	}
	errs, err := listenCmdStream(b.logSocketNamed(rt.Host, cmdErrSocket), spec.Stderr)
	if err != nil {
		out.Close()
		return nil, fmt.Errorf("stderr listener for %s: %w", rt.Host, err)
	}
	fail := func(err error) (CmdUnit, error) {
		out.Close()
		errs.Close()
		return nil, err
	}
	argv := append([]string{b.liveLink(rt.Host)}, spec.Args...)
	env := childEnv(b.cmdVars(rt, spec.genEnv, p.dbPath, spec.RunID, spec.RunDir))
	if err := b.writeLaunch(p, instancectl.RoleCmd, rt.Host, argv, env); err != nil {
		return fail(err)
	}
	var inv string
	if exeStart != nil {
		inv, err = exeStart.CmdExe(ctx, rt.Host, b.tier(rt).MemoryMax, *spec.serving)
	} else {
		inv, err = b.Control.Cmd(ctx, rt.Host, b.tier(rt).MemoryMax)
	}
	if err != nil {
		// The request may have reached systemd before it failed (a timeout,
		// a lost response, a 502 mid-roll), so the unit may be running.
		// Hand back a unit whose StopAndSettle stops the host's cmd role
		// and waits for its cgroup; it needs no invocation.
		return &cmdUnit{b: b, rt: rt, out: out, errs: errs}, fmt.Errorf("%s: start cmd unit: %w", rt.Host, err)
	}
	log.Printf("activator: %s: started %s for run %s", rt.Host, instancectl.UnitName(instancectl.RoleCmd, rt.Host), spec.RunID)
	return &cmdUnit{b: b, rt: rt, inv: inv, out: out, errs: errs}, nil
}

type cmdUnit struct {
	b         *ExecBackend
	rt        registry.Route
	inv       string
	out, errs *cmdStream
	mu        sync.Mutex // serialises StopAndSettle's attempts
	proven    bool       // an attempt succeeded; under mu
}

func (u *cmdUnit) Invocation() string { return u.inv }

func (u *cmdUnit) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
	return u.b.Control.CmdStatus(ctx, u.rt.Host, u.inv)
}

// StopAndSettle is one attempt at the proof: the cmd role stopped, then
// its cgroup waited empty, then both streams read to their end. The
// streams are drained only once the unit is proven gone, so a unit that
// outlives a failed attempt keeps its output flowing to the next one.
func (u *cmdUnit) StopAndSettle(ctx context.Context) error {
	u.mu.Lock()
	defer u.mu.Unlock()
	if u.proven {
		return nil
	}
	sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
	stopErr := u.b.Control.Stop(sctx, u.rt.Host, instancectl.RoleCmd)
	cancel()
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
	emptyErr := u.b.waitRoleCgroupEmpty(wctx, u.rt, instancectl.RoleCmd)
	wcancel()
	if err := errors.Join(stopErr, emptyErr); err != nil {
		return err
	}
	if err := errors.Join(u.out.Drain(cmdDrainBound), u.errs.Drain(cmdDrainBound)); err != nil {
		return err
	}
	u.proven = true
	return nil
}

// cmdStream is the agent's end of one cmd stream (run spec §5.5): the
// unit's StandardOutput=file: or StandardError=file: connects once, at
// exec, and every byte is copied to w unchanged until EOF. Nothing goes to
// the journal or the app log. One connection is taken; a second is closed
// unread.
type cmdStream struct {
	ln       net.Listener
	w        io.Writer
	accepted atomic.Bool
	closed   atomic.Bool
	mu       sync.Mutex
	conn     net.Conn
	copied   chan struct{}
	looped   chan struct{} // closed when the accept loop has returned
	wg       sync.WaitGroup
}

func listenCmdStream(path string, w io.Writer) (*cmdStream, error) {
	if w == nil {
		w = io.Discard
	}
	ln, err := edge.SecureUnixListen(path)
	if err != nil {
		return nil, err
	}
	s := &cmdStream{ln: ln, w: w, copied: make(chan struct{}), looped: make(chan struct{})}
	go s.accept()
	return s, nil
}

func (s *cmdStream) accept() {
	defer close(s.looped)
	for {
		c, err := s.ln.Accept()
		if err != nil {
			return
		}
		s.take(c)
	}
}

// take starts copying c if it is the stream's first connection, and closes
// it unread otherwise.
func (s *cmdStream) take(c net.Conn) {
	if s.accepted.Swap(true) {
		c.Close()
		return
	}
	s.mu.Lock()
	if s.closed.Load() {
		s.mu.Unlock()
		c.Close()
		return
	}
	s.conn = c
	s.wg.Add(1)
	s.mu.Unlock()
	go s.copy(c)
}

func (s *cmdStream) copy(c net.Conn) {
	defer s.wg.Done()
	defer close(s.copied)
	defer c.Close()
	_, _ = io.Copy(s.w, c)
}

// cmdAcceptGrace is how long Drain waits for a connection still queued in
// the listener's backlog. A var so tests can shrink it.
var cmdAcceptGrace = time.Second

// Drain waits up to bound for the stream to reach EOF (its unit has
// stopped, so every writer has closed), then closes the listener. The unit
// connected at exec, so the connection may still be queued in the backlog
// when the accept loop has not yet run: Drain stops the loop and takes any
// queued connection itself, so the command's last bytes are never dropped.
// A stream that never connected has nothing to drain.
func (s *cmdStream) Drain(bound time.Duration) error {
	if dl, ok := s.ln.(interface{ SetDeadline(time.Time) error }); ok {
		_ = dl.SetDeadline(time.Now())
		<-s.looped
		if !s.accepted.Load() {
			_ = dl.SetDeadline(time.Now().Add(cmdAcceptGrace))
			if c, err := s.ln.Accept(); err == nil {
				s.take(c)
			}
		}
	}
	var err error
	if s.accepted.Load() {
		select {
		case <-s.copied:
		case <-time.After(bound):
			err = fmt.Errorf("a cmd stream did not reach its end within %s", bound)
		}
	}
	s.Close()
	return err
}

func (s *cmdStream) Close() error {
	if s.closed.Swap(true) {
		return nil
	}
	err := s.ln.Close()
	s.mu.Lock()
	if s.conn != nil {
		s.conn.Close()
	}
	s.mu.Unlock()
	<-s.looped
	s.wg.Wait()
	return err
}