package host

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/runs"
)

// runPollRetry paces a failed long-poll; runRecheckEvery re-reads the
// record while a run executes, a backstop for a cancel whose serial bump
// this box missed.
var (
	runPollRetry    = time.Second
	runRecheckEvery = 30 * time.Second
	// runInterruptEvery is how often follow asks the handle whether an
	// instance stop or a retired generation interrupted the run, and
	// runInterruptGrace how long it then waits for the unit's own done
	// status, which wins if the command had already finished (re-review 6).
	runInterruptEvery = 250 * time.Millisecond
	runInterruptGrace = 2 * time.Second
)

// runOutputGrace bounds the output once the command has ended: the unit's
// stop (whose streams drain into the chunker) and the last chunk. The
// chunker's context outlives the command, so a cancel or a timeout never
// gives up the last chunk; this bound is what keeps a store that refuses
// every Put from retrying under the chunker's lock forever. It is longer
// than runStopBound.
var runOutputGrace = 5 * time.Minute

// runFinishBound bounds the retries of result.json and of the done CAS.
var runFinishBound = 2 * time.Minute

const (
	// runStopBound bounds CmdHandle.Stop: above the activator's own stop
	// bound, which covers the start, the stop, the cgroup and both streams.
	runStopBound = 3 * time.Minute
	// runReadBound bounds one read of the record.
	runReadBound = 30 * time.Second
)

var (
	errRunCancelledBeforeLaunch = errors.New("host: the run was cancelled before launch")
	errRunImportAtLaunch        = errors.New("host: the instance is an import that is not committed")
	errRunIncarnation           = errors.New("host: the record is another incarnation")
	errRunNotOurs               = errors.New("host: the record no longer names this run as this box's")
)

// exitOf reads systemd's ExecMainCode and ExecMainStatus. `systemctl show`
// prints the code as a number (1 exited, 2 killed, 3 dumped); the words
// are accepted too. A signal leaves exit 0 and names the signal.
func exitOf(st instancectl.CmdStatus) (int, string) {
	n, _ := strconv.Atoi(st.Status)
	switch st.Code {
	case "2", "3", "killed", "dumped":
		return 0, strconv.Itoa(n)
	}
	return n, ""
}

// runResultForLaunch maps RunCmd's refusals, and Prepare's, to the result
// the member reads. A route that is not an instance's (a worker) is
// refused as a flat box is: runs need an instance on the nested layout.
// Anything else (a busy hold, a failed download, a failed start) is the
// box failing to start the command.
func runResultForLaunch(err error) runs.Result {
	var we *activator.CmdWakeError
	var se *activator.CmdStoppedError
	switch {
	case errors.Is(err, activator.ErrCmdClaim):
		return runs.Result{Refused: runs.RefusedClaim}
	case errors.Is(err, activator.ErrCmdNotNested), errors.Is(err, activator.ErrCmdNotInstance):
		return runs.Result{Refused: runs.RefusedFlat}
	case errors.As(err, &we):
		return runs.Result{Refused: runs.RefusedWake, Detail: we.Reason}
	case errors.As(err, &se):
		return runs.Result{Interrupted: se.Reason}
	case errors.Is(err, errRunCancelledBeforeLaunch):
		return runs.Result{Cancelled: true}
	case errors.Is(err, errRunImportAtLaunch):
		return runs.Result{Refused: runs.RefusedImport}
	case errors.Is(err, errRunIncarnation):
		return runs.Result{Refused: runs.RefusedIncarnation}
	}
	return runs.Result{Interrupted: runs.InterruptedStartFailed}
}

// recordIncarnation is the incarnation request.json must carry for rec;
// "" when it cannot be computed, which matches no request.
func recordIncarnation(rec provision.Record) string {
	inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	if err != nil {
		return ""
	}
	return inc
}

// runFinishOurs is RunFinish for a run the record names as picked by box
// under boot. RunFinish itself does not check who picked the run.
func runFinishOurs(id, box, boot string) func(*provision.Record) error {
	finish := provision.RunFinish(id)
	return func(rec *provision.Record) error {
		if r := rec.ActiveRun; r == nil || r.ID != id || r.Box != box || r.BootID != boot {
			return fmt.Errorf("%w: the record does not name run %s as this box's", provision.ErrRunConflict, id)
		}
		return finish(rec)
	}
}

// execute is one picked run's life (run spec §5.2 to §5.7): the
// incarnation, the launch (Prepare checks the record again and stages the
// files), the poll, a cancel or a timeout, then the unit stopped, every
// chunk written, result.json, the done CAS, the run directory deleted and
// the hold released, in that order.
func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
	defer w.forget(req.ID)
	instDir := provision.InstanceDir(w.dataDir, rec.Host)
	if req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host || req.Incarnation == "" || req.Incarnation != recordIncarnation(rec) {
		w.finish(ctx, rec, req, runs.Result{Refused: runs.RefusedIncarnation}, nil)
		return
	}
	// The chunker's context is not the run's: a cancel or a timeout must
	// not give up the last chunk. endOutput is armed once the command has
	// ended (runOutputGrace).
	octx, endOutput := context.WithCancel(context.WithoutCancel(ctx))
	defer endOutput()
	chunks := runs.NewChunker(octx, w.store, rec.Account, rec.App, req.ID)
	closeOutput := func(res *runs.Result) {
		last, truncated, err := chunks.Close()
		if err != nil {
			log.Printf("agent: runs: %s %s: output: %v", rec.Host, req.ID, err)
		}
		res.LastSeq, res.Truncated = last, truncated
	}
	// notOurs is prepare's own verdict: RunCmd reports a stop that
	// overlapped Prepare in place of Prepare's error, and a run the record
	// no longer gives this box must get no result whatever RunCmd says.
	notOurs := false
	// prepRefused: Prepare answered an error, so its own result, not the
	// launch context's end, is the run's story (a staging that meets the
	// deadline stays start failed). recheckErr is the post-switch
	// Recheck's answer: a refusal keeps its own result too, but a record
	// read the run's own cancel or deadline cut short is not a refusal
	// (re-review 7).
	prepRefused := false
	var recheckErr error
	spec := activator.CmdSpec{
		RunID:  req.ID,
		Args:   req.Argv,
		RunDir: filepath.Join(instDir, runs.DirName, req.ID),
		Stdout: chunks.Stream(runs.StreamOut),
		Stderr: chunks.Stream(runs.StreamErr),
		Prepare: func(ctx context.Context) error {
			err := w.prepare(ctx, instDir, req, h, launchDeadline(rec, req, w.now()))
			notOurs, prepRefused = errors.Is(err, errRunNotOurs), err != nil
			return err
		},
		Recheck: func(ctx context.Context) error {
			err := w.launchable(ctx, req, h)
			notOurs, recheckErr = errors.Is(err, errRunNotOurs), err
			return err
		},
	}
	// The launch runs under the run's deadline and its cancel, the whole
	// admission through (re-review 5): the wake, Prepare, a wait for a
	// swap's switch and the last checks. A run cancelled or expired at
	// any of them starts nothing.
	lctx, lcancel := w.launchContext(ctx, rec, req, h)
	handle, err := w.launch.RunCmd(lctx, rt, spec)
	launchEnded := lctx.Err()
	lcancel()
	if err != nil {
		bound := time.AfterFunc(runOutputGrace, endOutput)
		defer bound.Stop()
		if notOurs || errors.Is(err, errRunNotOurs) {
			// The record gives the run to no one, or to another boot:
			// whoever holds it writes its result.
			chunks.Close()
			w.cleanUp(rec, req, nil)
			log.Printf("agent: runs: %s %s: not launched: %v", rec.Host, req.ID, err)
			return
		}
		res := runResultForLaunch(err)
		recheckRefused := recheckErr != nil && !(launchEnded != nil && endedByContext(recheckErr))
		if launchEnded != nil && !prepRefused && !recheckRefused {
			// The admission ended with the run's own cancel or deadline:
			// the run's own result, as after launch (re-review 5).
			cancelled := cancelRequested(h)
			res = runs.Result{Cancelled: cancelled, TimedOut: !cancelled}
		}
		if res.Interrupted == runs.InterruptedStartFailed {
			log.Printf("agent: runs: %s %s: start: %v", rec.Host, req.ID, err)
		}
		closeOutput(&res)
		w.finish(ctx, rec, req, res, nil)
		return
	}
	log.Printf("agent: runs: %s: run %s started (%s)", rec.Host, req.ID, handle.Invocation())
	w.runEvent(rec, runStartedLine(req))
	res := runs.Result{Started: w.now().UTC()}
	status, doneAt := w.follow(ctx, rec, req, handle, h, &res)
	bound := time.AfterFunc(runOutputGrace, endOutput)
	defer bound.Stop()
	// The unit is stopped and settled after every ending, done included
	// (run spec §5.4), so its streams have reached their end before the
	// chunker closes and the result names the last chunk.
	sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runStopBound)
	if err := handle.Stop(sctx); err != nil {
		log.Printf("agent: runs: %s %s: stop: %v", rec.Host, req.ID, err)
	}
	cancel()
	if status == nil && (res.Cancelled || res.TimedOut) {
		pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineCmdPoll)
		if st, err := handle.Poll(pctx); err == nil && st.State == instancectl.CmdDone {
			status = &st
		}
		pcancel()
	}
	if status != nil && status.State == instancectl.CmdDone {
		res.Exit, res.Signal = exitOf(*status)
	}
	reason, askedAt := handle.Interrupted()
	if reason != "" && status != nil && status.State == instancectl.CmdDone && !doneAt.IsZero() && askedAt.After(doneAt) {
		// The instance stop was asked for after the done status arrived:
		// the command had finished on its own, and the done status wins.
		reason = ""
	}
	if reason != "" {
		res.Interrupted = reason
	} else if status != nil && status.State == instancectl.CmdGone && !res.Cancelled && !res.TimedOut {
		res.Interrupted = runs.InterruptedGone
	}
	closeOutput(&res)
	w.finish(ctx, rec, req, res, handle)
}

// endedByContext reports whether err is a context's end: a read cut short
// by a cancel or a deadline, never a refusal on the record's say.
func endedByContext(err error) bool {
	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
}

// launchContext is ctx bounded by the run's deadline and ended by its
// cancel: what RunCmd runs under (re-review 5). The cancel func must be
// called once RunCmd has returned.
func (w *runsWorker) launchContext(ctx context.Context, rec provision.Record, req runs.Request, h *runHandle) (context.Context, context.CancelFunc) {
	lctx, cancel := context.WithDeadline(ctx, launchDeadline(rec, req, w.now()))
	go func() {
		select {
		case <-h.cancel:
			cancel()
		case <-lctx.Done():
		}
	}()
	return lctx, cancel
}

// launchDeadline is the run's deadline from the record (RunPick set it to
// the pickup plus the timeout), or now plus the timeout without one.
func launchDeadline(rec provision.Record, req runs.Request, now time.Time) time.Time {
	if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() {
		return r.Deadline
	}
	return now.Add(req.Timeout())
}

// prepare is the run's CmdSpec.Prepare, run under the activator's hold
// right before the unit starts (run spec §5.2): the record must still name
// the run as this box's under this boot, not cancelled, not an uncommitted
// import, the same incarnation. Then the files are staged, a cancel cutting
// the download short, and the record is checked once more, so a cancel
// that lands while the files download still starts no unit. All of it ends
// at the run's deadline: a stalled download must not keep the hold (and
// with it every later run on the instance) forever.
func (w *runsWorker) prepare(ctx context.Context, instDir string, req runs.Request, h *runHandle, deadline time.Time) error {
	ctx, cancel := context.WithDeadline(ctx, deadline)
	defer cancel()
	if err := w.launchable(ctx, req, h); err != nil {
		return err
	}
	sctx, stop := context.WithCancel(ctx)
	defer stop()
	go func() {
		select {
		case <-h.cancel:
			stop()
		case <-sctx.Done():
		}
	}()
	if _, err := runs.StageFiles(sctx, w.store, instDir, req); err != nil {
		if cancelRequested(h) {
			return errRunCancelledBeforeLaunch
		}
		return fmt.Errorf("host: stage the run's files: %w", err)
	}
	return w.launchable(ctx, req, h)
}

// launchable reads the record and says why the run must not start, or nil.
func (w *runsWorker) launchable(ctx context.Context, req runs.Request, h *runHandle) error {
	if cancelRequested(h) {
		return errRunCancelledBeforeLaunch
	}
	rctx, cancel := context.WithTimeout(ctx, runReadBound)
	defer cancel()
	cur, err := provision.GetRecord(rctx, w.store, req.Account, req.App, req.Host)
	if errors.Is(err, os.ErrNotExist) {
		return errRunNotOurs
	}
	if err != nil {
		return fmt.Errorf("host: read the record before launch: %w", err)
	}
	r := cur.ActiveRun
	switch {
	case r == nil || r.ID != req.ID || r.Box != w.box || r.BootID != w.boot || (r.State != provision.RunPicked && r.State != provision.RunCancel):
		return errRunNotOurs
	case r.State == provision.RunCancel:
		return errRunCancelledBeforeLaunch
	case cur.ImportUncommitted():
		return errRunImportAtLaunch
	case recordIncarnation(*cur) != req.Incarnation:
		return errRunIncarnation
	}
	return nil
}

func cancelRequested(h *runHandle) bool {
	select {
	case <-h.cancel:
		return true
	default:
		return false
	}
}

// follow long-polls the unit until it is done or gone, the run is
// cancelled, or its timeout passes. It returns the final status when the
// unit reported one, with when the poll answered it, and marks a cancel or
// a timeout on res.
func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) (*instancectl.CmdStatus, time.Time) {
	pctx, stopPolling := context.WithCancel(ctx)
	defer stopPolling()
	type polled struct {
		st instancectl.CmdStatus
		at time.Time
	}
	statuses := make(chan polled, 1)
	go func() {
		for {
			st, err := handle.Poll(pctx)
			if pctx.Err() != nil {
				return
			}
			if err != nil {
				select {
				case <-pctx.Done():
					return
				case <-time.After(runPollRetry):
				}
				continue
			}
			if st.State != instancectl.CmdRunning {
				statuses <- polled{st, w.now()}
				return
			}
		}
	}()
	// The timeout is the run's own deadline, set at pickup: staging that
	// ran long leaves the command less, not a full timeout from launch.
	limit := req.Timeout()
	if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() {
		limit = min(limit, r.Deadline.Sub(w.now()))
	}
	timeout := time.NewTimer(limit)
	defer timeout.Stop()
	recheck := time.NewTicker(runRecheckEvery)
	defer recheck.Stop()
	interrupts := time.NewTicker(runInterruptEvery)
	defer interrupts.Stop()
	for {
		select {
		case p := <-statuses:
			return &p.st, p.at
		case <-interrupts.C:
			// An interruption ends the run as interrupted (exit 125) at
			// once, even while the activator is still retrying the unit's
			// stop (re-review 6): the command's result does not wait on a
			// stop that may take a while to be proven. A done status that
			// arrives within the grace is still read, so a command that had
			// finished first keeps its exit.
			if reason, _ := handle.Interrupted(); reason != "" {
				grace := time.NewTimer(runInterruptGrace)
				defer grace.Stop()
				select {
				case p := <-statuses:
					return &p.st, p.at
				case <-grace.C:
					return nil, time.Time{}
				}
			}
		case <-h.cancel:
			res.Cancelled = true
			return nil, time.Time{}
		case <-timeout.C:
			res.TimedOut = true
			return nil, time.Time{}
		case <-recheck.C:
			rctx, cancel := context.WithTimeout(ctx, runReadBound)
			cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host)
			cancel()
			if err == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == req.ID && cur.ActiveRun.State == provision.RunCancel {
				h.requestCancel()
			}
		}
	}
}

// finish writes the run's result, then CASes the run to done only if the
// result landed and the record names the run as this box's under this
// boot, then cleans up. A result that cannot be written leaves the run
// picked: the sweep ends it as lost with a result of its own, so a run is
// never done without one.
func (w *runsWorker) finish(ctx context.Context, rec provision.Record, req runs.Request, res runs.Result, handle activator.CmdHandle) {
	res.Box, res.Ended = w.box, w.now().UTC()
	fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runFinishBound)
	defer cancel()
	err := untilLanded(fctx, func(ctx context.Context) error {
		return runs.WriteResult(ctx, w.store, rec.Account, rec.App, req.ID, res)
	})
	if err != nil {
		log.Printf("agent: runs: %s %s: result: %v", rec.Host, req.ID, err)
	} else {
		// The ending is logged only once the result has landed (amadan
		// #325: run lines are at most once). A result that never landed
		// leaves the run picked, and a later agent's closeInterrupted
		// writes and logs its own ending; logging this one too would put
		// two contradictory endings in the log. A run with no handle never
		// had its unit started.
		w.runEvent(rec, runEndedLine(req.ID, res, handle != nil))
		if err := untilLanded(fctx, func(ctx context.Context) error {
			_, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, runFinishOurs(req.ID, w.box, w.boot))
			return err
		}); err != nil && !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("agent: runs: %s %s: done: %v", rec.Host, req.ID, err)
		}
	}
	w.cleanUp(rec, req, handle)
}

// cleanUp deletes the run's directory (an error is logged, never acted
// on) and releases the activator's hold.
func (w *runsWorker) cleanUp(rec provision.Record, req runs.Request, handle activator.CmdHandle) {
	if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), req.ID); err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, req.ID, err)
	}
	if handle != nil {
		handle.Release()
	}
}

// untilLanded retries f until it succeeds, answers a conflict or a missing
// object (neither changes on a retry), or ctx ends.
func untilLanded(ctx context.Context, f func(context.Context) error) error {
	backoff := 250 * time.Millisecond
	for {
		err := f(ctx)
		if err == nil || errors.Is(err, provision.ErrRunConflict) || errors.Is(err, os.ErrNotExist) {
			return err
		}
		select {
		case <-ctx.Done():
			return err
		case <-time.After(backoff):
		}
		backoff = min(backoff*2, 5*time.Second)
	}
}