package activator

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync/atomic"
	"time"

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

// ErrStillRunning is Stop's answer when ctx expired before termination was
// proven (design §6.3). The activator then keeps the lease and the entry
// in stopping, and the sweeper retries; the lease is never released over a
// writer that may still exist.
var ErrStillRunning = errors.New("activator: stop unproven; processes remain")

// unitProc is a nested host's Proc (design §6.3): the unit the control
// service started, the listeners the agent holds for it, and the observer
// that proves termination. Named apart from backend_unit.go's svcUnitProc,
// a whole-app tenant's own systemd unit: this one is a nested exec
// instance the activator spawns through the root instance-control channel.
type unitProc struct {
	b           *ExecBackend
	rt          registry.Route
	role        instancectl.Role
	unit, slice string
	invocation  string
	runDir      string
	// gen is the generation of a carlos-instance-gen@ unit, 0 for the
	// original templates; dataDir holds its launch file.
	gen       uint64
	dataDir   string
	ctl       *edge.ControlListener
	logs      *logListener
	exited    atomic.Bool
	done      chan struct{}
	closeOnce atomic.Bool
	// listenersArmed guards the done-watcher goroutine Stop's ctx.Done
	// branch starts: several callers bound their wait on Stop shorter than
	// the unit may actually take to die (a sidecar co-stop, a discarded
	// sidecar, a failed health wait) and never call Stop again once it
	// returns ErrStillRunning, which otherwise leaves ctl and logs open
	// for as long as the agent runs. Armed at most once per unitProc.
	listenersArmed atomic.Bool
	// stopRequested is set before the FIRST Control.Stop this proc's owner
	// ever issues for it: once true, the observer stops treating a
	// differently-invoked or not-yet-invoked unit as a FOREIGN restart to
	// stop on its own. It is instead the proof that our own invocation is
	// already over (see observe).
	stopRequested atomic.Bool
	// genFilesGone is set once a stop of this generation succeeded: the
	// control service removed the unit's drop-in and exe layout with it,
	// so finaliseGen need not ask again.
	genFilesGone atomic.Bool
	// launchEnv is the file environment (the config bundle with the
	// instance's own file on top) a run role's unit launched with, before
	// the agent-owned variables. It is written once, before the handle
	// leaves spawnUnitFrom, and never again. A run against this generation
	// composes its environment from it (LaunchEnv; merge ruling M3). nil
	// for every other role.
	launchEnv map[string]string
	// launchKeyLayer is the layer launchEnv's CARLOS_INSTANCE_KEY came
	// from, decided by the read that composed launchEnv (instance-key
	// ruling 29). Written with launchEnv, never again.
	launchKeyLayer instancekey.Layer
}

// launchEnvProc is a generation handle that kept the file environment its
// generation launched with (merge ruling M3).
type launchEnvProc interface {
	LaunchEnv() map[string]string
}

// LaunchEnv is the file environment this generation launched with, or nil
// when it kept none. The map is the handle's own: callers must not write
// it.
func (p *unitProc) LaunchEnv() map[string]string { return p.launchEnv }

// launchKeyLayerProc is a generation handle that also kept where its
// launch environment's instance key came from.
type launchKeyLayerProc interface {
	LaunchKeyLayer() instancekey.Layer
}

// LaunchKeyLayer is the layer LaunchEnv's CARLOS_INSTANCE_KEY came from.
func (p *unitProc) LaunchKeyLayer() instancekey.Layer { return p.launchKeyLayer }

func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc {
	return b.newUnitProcFor(rt, role, 0, p, inv, ctl, logs)
}

func (b *ExecBackend) newUnitProcFor(rt registry.Route, role instancectl.Role, gen uint64, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc {
	up := &unitProc{b: b, rt: rt, role: role, gen: gen, unit: unitName(role, rt.Host, gen), slice: instancectl.SliceName(rt.Host),
		invocation: inv, runDir: p.runDir, dataDir: p.dataDir, ctl: ctl, logs: logs, done: make(chan struct{})}
	go b.observe(up)
	return up
}

// unitName is the unit a role of host runs as: the original template's, or
// generation gen's (zero-downtime spec §7.6).
func unitName(role instancectl.Role, host string, gen uint64) string {
	if gen == 0 {
		return instancectl.UnitName(role, host)
	}
	return instancectl.GenUnitName(host, gen)
}

// stopRole stops one role's unit: the original template's through Control,
// a generation's through GenControl. A generation stop never touches a
// sibling generation.
func (b *ExecBackend) stopRole(ctx context.Context, host string, role instancectl.Role, gen uint64) error {
	if gen == 0 {
		return b.Control.Stop(ctx, host, role)
	}
	gc, err := b.genControl()
	if err != nil {
		return err
	}
	return gc.StopGen(ctx, host, gen, false)
}

func (p *unitProc) Exited() bool { return p.exited.Load() }

// Stop asks the control service to stop the unit and waits for termination
// to be proven. The control listener and the log listener are closed only
// after that, preserving childProc's ordering (signal first, close the
// listener after the process is gone).
func (p *unitProc) Stop(ctx context.Context) error {
	if !p.Exited() {
		// Set before the Control.Stop call, not after: observe polls
		// concurrently, and a stop it cannot yet see as "requested" would
		// still fire its own foreign-invocation stop for the same unit.
		p.stopRequested.Store(true)
		sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl)
		err := p.stopUnit(sctx)
		cancel()
		if err != nil {
			log.Printf("activator: stop %s %s: %v", p.rt.Host, p.role, err)
		}
	}
	select {
	case <-p.done:
		p.closeListeners()
		return nil
	case <-ctx.Done():
		// The caller is giving up on waiting, not on the stop itself: the
		// unit may still be mid-drain, so the listeners must stay open while
		// a writer could still exist. But several callers never call Stop
		// again, and closing them still has to happen once done eventually
		// closes, or they leak for the life of the agent.
		// One goroutine per unitProc, armed once, does that regardless of
		// whether anyone ever retries.
		if !p.listenersArmed.Swap(true) {
			go func() {
				<-p.done
				p.closeListeners()
			}()
		}
		return ErrStillRunning
	}
}

// stopUnit asks the control service to stop this proc's unit. For a
// generation, a stop that succeeded also removed the unit's files there.
func (p *unitProc) stopUnit(ctx context.Context) error {
	err := p.b.stopRole(ctx, p.rt.Host, p.role, p.gen)
	if err == nil && p.gen != 0 {
		p.genFilesGone.Store(true)
	}
	return err
}

// finaliseGen removes a generation's files once observe has proven it
// gone (generation numbers are never reused, so nothing else would): its
// launch file here, and the control service's drop-in and exe layout
// through a stop of the inactive unit, unless a stop of ours already
// removed them. Never called while termination is unproven.
func (p *unitProc) finaliseGen() {
	if p.gen == 0 {
		return
	}
	p.removeLaunch()
	if p.genFilesGone.Load() {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	if err := p.stopUnit(ctx); err != nil {
		log.Printf("activator: %s: remove the files of %s: %v", p.rt.Host, p.unit, err)
	}
}

func (p *unitProc) closeListeners() {
	if p.closeOnce.Swap(true) {
		return
	}
	if p.ctl != nil {
		p.ctl.Close()
	}
	if p.logs != nil {
		p.logs.Close()
	}
}

// observe is the goroutine of design §6.5. It proves termination from two
// facts the tenant cannot forge: systemd's unit state and cgroup emptiness.
// The exit record decides the wording of the event line and nothing else.
func (b *ExecBackend) observe(p *unitProc) {
	poll := b.ObservePoll
	if poll <= 0 {
		poll = time.Second
	}
	state := b.UnitState
	if state == nil {
		state = systemctlUnitState
	}
	events := b.CgroupEvents
	if events == nil {
		events = cgroupEventsUnder(b.cgroupRoot())
	}
	foreignStopped := false
	lastRemain := time.Time{}
	lastEventsErr := time.Time{}
	for {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		active, inv, err := state(ctx, p.unit)
		cancel()
		if err != nil {
			log.Printf("activator: %s: unit state: %v", p.rt.Host, err)
		}
		terminal := false
		foreign := false
		switch {
		case err != nil:
		case inv != "" && inv != p.invocation:
			foreign = true
		case active == "inactive" || active == "failed":
			terminal = true
		}

		// A stop of OUR OWN invocation is outstanding (Stop was already
		// called): systemd starts a fresh invocation of the same unit only
		// once the previous one's stop job has fully completed, so seeing
		// anything that is not provably that same old invocation, a
		// different id, or none reported yet while the unit sits in a
		// non-terminal ActiveState (mid-cycle to a fresh one), is itself
		// proof our invocation is over. No cgroup check is needed here:
		// there is no "our process might still be running" left to
		// disprove, because systemd's own state machine already disproved
		// it by allowing a new invocation to begin.
		if p.stopRequested.Load() && err == nil && !terminal && (foreign || inv == "") {
			emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: replaced after stop)")
			p.exited.Store(true)
			close(p.done)
			return
		}

		if foreign && !foreignStopped && !p.stopRequested.Load() {
			log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation)
			sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
			if err := p.stopUnit(sctx); err != nil {
				log.Printf("activator: %s: stop foreign invocation: %v", p.rt.Host, err)
			}
			scancel()
			foreignStopped = true
		}
		if terminal || (foreign && foreignStopped) {
			exists, populated, eerr := events(p.slice, p.unit)
			if eerr != nil {
				// A read error is not evidence of anything. Not exited,
				// not still running, just unknown. Treating it as "cgroup
				// empty" would let a still-running process's exit be
				// reported anyway, exactly what this proof scheme exists
				// to prevent (design §6.5); rate-limited the same way the
				// "processes remain" line below is, since a persistent
				// read failure would otherwise flood the log every poll.
				if time.Since(lastEventsErr) > 2*time.Second {
					log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr)
					lastEventsErr = time.Now()
				}
				time.Sleep(poll)
				continue
			}
			if !exists || !populated {
				if terminal {
					rec, ok := p.readRecord()
					emitEvent(b.Logs, p.rt, exitLevel(rec, ok), exitLine(rec, ok))
					// The record is single-use (design §6.5): consumed
					// exactly once into the member-facing line above, then
					// cleared so it can never be misread as evidence for a
					// LATER invocation of this same role (prepare only
					// clears records left over from EARLIER invocations;
					// this one is this invocation's own to remove).
					p.removeRecord()
				} else {
					emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)")
				}
				p.finaliseGen()
				p.exited.Store(true)
				// Stop closes the listeners when it observes done; a crash
				// nobody is stopping yet leaves them for the sweeper's
				// stand-down, which calls Stop.
				close(p.done)
				return
			}
			if time.Since(lastRemain) > 2*time.Second {
				log.Printf("activator: %s: processes remain after %s; not releasing", p.rt.Host, active)
				lastRemain = time.Now()
			}
		}
		time.Sleep(poll)
	}
}

// exitLine words the event line from the advisory record observe already
// read once and passes to both exitLine and exitLevel, so the two can
// never disagree from reading the file twice with something changing in
// between, the same way childProc does: a non-zero status is a warn line,
// a clean exit is info, and no record is unknown. The record is
// tenant-writable text (design §6.5), so every field that reaches the line
// passes through recordToken first.
func exitLine(rec instancectl.ExitRecord, ok bool) string {
	switch {
	case !ok:
		return "instance exited (status unknown: no exit record)"
	case rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0"):
		return "instance exited (exit 0)"
	case rec.Code == "killed" || rec.Code == "dumped":
		return "instance exited: signal " + recordToken(rec.Status)
	default:
		return fmt.Sprintf("instance exited: exit %s (%s)", recordToken(rec.Status), recordToken(rec.Result))
	}
}

// recordToken bounds a field of the exit record to what systemd itself
// would write: at most 32 bytes of [A-Za-z0-9_-]. Anything else becomes
// "?" so a tenant cannot put arbitrary text or control characters into
// the platform's event stream.
func recordToken(s string) string {
	if len(s) == 0 || len(s) > 32 {
		return "?"
	}
	for _, c := range s {
		if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-') {
			return "?"
		}
	}
	return s
}

// exitLevel is exitLine's companion, from the same read (see exitLine).
func exitLevel(rec instancectl.ExitRecord, ok bool) string {
	if ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) {
		return "info"
	}
	return "warn"
}

func (p *unitProc) recordName() string {
	if p.gen == 0 {
		return instancectl.ExitRecordName(p.role, p.invocation)
	}
	return instancectl.GenExitRecordName(p.gen, p.invocation)
}

func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) {
	raw, err := instancefs.ReadBeneathLimit(p.runDir, p.recordName(), 4096)
	if err != nil {
		return instancectl.ExitRecord{}, false
	}
	var rec instancectl.ExitRecord
	if err := json.Unmarshal(raw, &rec); err != nil {
		return instancectl.ExitRecord{}, false
	}
	return rec, true
}

// removeRecord deletes this invocation's own advisory exit record (design
// §6.5) after observe has consumed it into the member-facing event line.
// Best-effort: a record that is already gone, or a filesystem that refuses
// the remove, leaves nothing worse than what prepare's own stale-record
// sweep would have cleared on the NEXT invocation anyway.
func (p *unitProc) removeRecord() {
	_ = os.Remove(filepath.Join(p.runDir, p.recordName()))
}

// removeLaunch deletes a generation's launch file once the generation is
// proven gone. Generation numbers are never reused, so nothing else would.
// The original templates' launch file is rewritten by every start and is
// left alone. os.Remove never follows the final component.
func (p *unitProc) removeLaunch() {
	if p.gen != 0 && p.dataDir != "" {
		_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(p.gen)))
	}
}

// cgroupRoot is CgroupRoot with the nil default applied.
func (b *ExecBackend) cgroupRoot() string {
	if b.CgroupRoot != "" {
		return b.CgroupRoot
	}
	return "/sys/fs/cgroup"
}