package host

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"

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

// runLauncher is what the runs worker needs from the activator.
type runLauncher interface {
	RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error)
}

var _ runLauncher = (*activator.Service)(nil)

// runHandle is the worker's line to one executing run.
type runHandle struct {
	cancel chan struct{}
	once   sync.Once
}

func newRunHandle() *runHandle { return &runHandle{cancel: make(chan struct{})} }

func (h *runHandle) requestCancel() { h.once.Do(func() { close(h.cancel) }) }

// runsWorker is the box's half of carlos run (run spec §5.2). boot is this
// agent process's own id: a run picked under another boot of this box lost
// its unit when the edge stopped (BindsTo=carlos-edge.service).
type runsWorker struct {
	store   release.Store
	routes  func(host string) (*registry.Route, error)
	launch  runLauncher
	dataDir string
	box     string
	boot    string
	now     func() time.Time
	exec    func(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle)
	// events is where a run's start and end reach `carlos logs`
	// (runevents.go); nil on a box with no log store.
	events activator.EventSink
	// start runs a picked run's exec: on its own goroutine in the agent,
	// inline in tests that count launches.
	start func(func())
	// booted is set when the boot pass has finished. The boot pass deletes
	// every .carlos-run directory, so until then pass does nothing: a run
	// it launched would have its directory swept from under it.
	booted atomic.Bool

	mu     sync.Mutex
	noted  map[string]provision.Record // host -> the record reconcile read
	active map[string]*runHandle       // run id -> the run this agent is executing
	unfit  map[string]bool             // run ids whose request.json was refused, logged once
}

func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker {
	w := &runsWorker{store: store, routes: routes, launch: launch, dataDir: dataDir, box: box, boot: boot,
		now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{}, unfit: map[string]bool{},
		start: func(f func()) { go f() }}
	w.exec = w.execute
	return w
}

// newBootID is a fresh id for this agent process.
func newBootID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		return time.Now().UTC().Format("20060102T150405.000000000")
	}
	return hex.EncodeToString(b[:])
}

// armedRuns is the worker the agent armed; nil until then (and on a box
// with no activator, which cannot run anything).
var armedRuns atomic.Pointer[runsWorker]

// note is ReconcileConfig.RunSeen: remember, never act, never block.
func (w *runsWorker) note(rec provision.Record) {
	w.mu.Lock()
	w.noted[rec.Host] = rec
	w.mu.Unlock()
}

// pass acts on every record noted since the last pass, once the boot pass
// has finished; before that the notes wait.
func (w *runsWorker) pass(ctx context.Context) {
	if !w.booted.Load() {
		return
	}
	w.mu.Lock()
	noted := w.noted
	w.noted = map[string]provision.Record{}
	w.mu.Unlock()
	for _, rec := range noted {
		w.consider(ctx, rec)
	}
}

// bootPass runs once at agent start (index item 2: reconcile's first pass
// is a minute away, and the CLI expires a run at 60 s). It deletes every
// .carlos-run directory first (run spec §5.8: no run is open on the box
// yet), then reads every record once and acts on each run.
func (w *runsWorker) bootPass(ctx context.Context) {
	// Opens pass, whether or not the records could be read: runs the boot
	// pass did not see are the next passes' to pick.
	defer w.booted.Store(true)
	if err := runs.SweepRunDirs(w.dataDir); err != nil {
		log.Printf("agent: runs: boot sweep of .carlos-run: %v", err)
	}
	recs, err := provision.ListAllRecords(ctx, w.store)
	if err != nil {
		log.Printf("agent: runs: boot pass: %v", err)
		return
	}
	for _, rec := range recs {
		if rec.ActiveRun != nil {
			w.consider(ctx, rec)
		}
	}
}

// owned is the row this box runs the record's instance from, when it is
// this box's to act on: provisioned for this record's account and app,
// and one the activator manages.
func (w *runsWorker) owned(rec provision.Record) (registry.Route, bool) {
	rt, err := w.routes(rec.Host)
	if err != nil || rt == nil || !rt.Provisioned || rt.Account != rec.Account || rt.App != rec.App || !activator.ActivatorManaged(*rt) {
		return registry.Route{}, false
	}
	return *rt, true
}

// consider acts on one record's run (run spec §5.2).
func (w *runsWorker) consider(ctx context.Context, rec provision.Record) {
	r := rec.ActiveRun
	if r == nil {
		return
	}
	switch r.State {
	case provision.RunQueued:
		rt, ok := w.owned(rec)
		if !ok || w.now().After(r.Deadline) {
			return
		}
		w.pickup(ctx, rec, rt)
	case provision.RunPicked, provision.RunCancel:
		if r.Box != w.box {
			return
		}
		w.mu.Lock()
		h := w.active[r.ID]
		w.mu.Unlock()
		switch {
		case h != nil && r.State == provision.RunCancel:
			h.requestCancel()
		case h == nil && r.BootID != w.boot:
			w.closeInterrupted(ctx, rec)
		case h == nil:
			w.finishStranded(ctx, rec)
		}
	}
}

// finishStranded is the done CAS again for a run this box picked under this
// boot and is no longer executing, whose result landed but whose done CAS
// did not within runFinishBound. Left picked, it would hold the instance
// until the sweep's deadline plus an hour. A run with no result is left
// alone: its result could not be written, and the sweep ends it as lost.
func (w *runsWorker) finishStranded(ctx context.Context, rec provision.Record) {
	id := rec.ActiveRun.ID
	rctx, cancel := context.WithTimeout(ctx, runReadBound)
	defer cancel()
	res, err := runs.ReadResult(rctx, w.store, rec.Account, rec.App, id)
	if err != nil || res == nil {
		return
	}
	if _, err := provision.SetRecordRun(rctx, w.store, rec.Account, rec.App, rec.Host, runFinishOurs(id, w.box, w.boot)); err != nil {
		if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("agent: runs: %s %s: done, again: %v", rec.Host, id, err)
		}
		return
	}
	log.Printf("agent: runs: %s: run %s had its result but was still picked; marked done", rec.Host, id)
}

// pickup takes a queued run with one CAS from queued to picked. Only the
// box whose CAS lands launches; any other outcome (another box, a cancel,
// an expiry) leaves the run alone. A request.json that is not the run the
// record names, or whose timeout is out of range, is never picked: the
// expiry ends it.
func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt registry.Route) {
	id := rec.ActiveRun.ID
	req, err := runs.ReadRequest(ctx, w.store, rec.Account, rec.App, id)
	if err != nil {
		// The console writes request.json before the record names the run;
		// without it there is nothing to run, and the expiry ends it.
		log.Printf("agent: runs: %s %s: request: %v", rec.Host, id, err)
		return
	}
	if err := requestFits(rec, id, *req); err != nil {
		if w.firstUnfit(id) {
			log.Printf("agent: runs: %s %s: not picked: %v", rec.Host, id, err)
		}
		return
	}
	h := newRunHandle()
	w.mu.Lock()
	if w.active[id] != nil {
		w.mu.Unlock()
		return
	}
	w.active[id] = h
	w.mu.Unlock()
	got, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host,
		provision.RunPick(id, w.box, w.boot, w.now().Add(req.Timeout())))
	if err != nil {
		if got = w.pickedAnyway(ctx, rec, id); got == nil {
			w.forget(id)
			if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
				log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err)
			}
			return
		}
		log.Printf("agent: runs: %s %s: pickup answered %v, but the record names the run as this box's", rec.Host, id, err)
	}
	log.Printf("agent: runs: %s: picked run %s", rec.Host, id)
	w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })
}

// unfitLogMax caps the ids firstUnfit remembers; past it the set starts
// again, and a refusal still queued is logged once more.
const unfitLogMax = 256

// firstUnfit reports whether this is the first refusal of id's request,
// so a run left for the expiry is logged once, not on every pass.
func (w *runsWorker) firstUnfit(id string) bool {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.unfit[id] {
		return false
	}
	if len(w.unfit) >= unfitLogMax {
		clear(w.unfit)
	}
	w.unfit[id] = true
	return true
}

// requestFits says why request.json is not the run rec names, with a
// timeout the console would have accepted, or nil.
func requestFits(rec provision.Record, id string, req runs.Request) error {
	if req.ID != id || req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host {
		return fmt.Errorf("request.json names %s/%s/%s %s, not the record's run", req.Account, req.App, req.Host, req.ID)
	}
	return runs.CheckTimeout(req.Timeout())
}

// pickedAnyway reads the record back after a pickup CAS that answered an
// error: a write whose answer was lost, or a conditional write that landed
// and then answered a conflict, may have picked the run all the same. The
// record is what the box launches from when it names the run as picked by
// this box under this boot; nil otherwise.
func (w *runsWorker) pickedAnyway(ctx context.Context, rec provision.Record, id string) *provision.Record {
	for range 3 {
		rctx, cancel := context.WithTimeout(ctx, runReadBound)
		cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host)
		cancel()
		if errors.Is(err, os.ErrNotExist) {
			return nil
		}
		if err == nil {
			if r := cur.ActiveRun; r != nil && r.ID == id && r.Box == w.box && r.BootID == w.boot &&
				(r.State == provision.RunPicked || r.State == provision.RunCancel) {
				return cur
			}
			return nil
		}
		log.Printf("agent: runs: %s %s: read back the pickup: %v", rec.Host, id, err)
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(runPollRetry):
		}
	}
	return nil
}

func (w *runsWorker) forget(id string) {
	w.mu.Lock()
	delete(w.active, id)
	w.mu.Unlock()
}

// closeInterrupted ends a run this box picked under an earlier boot (run
// spec §5.2): its unit went with the edge, so it is never executed again.
// The result names the last chunk the old boot wrote.
func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record) {
	id := rec.ActiveRun.ID
	last, err := runs.LastChunkSeq(ctx, w.store, rec.Account, rec.App, id)
	if err != nil {
		log.Printf("agent: runs: %s %s: list chunks: %v", rec.Host, id, err)
		return
	}
	res := runs.Result{Interrupted: runs.InterruptedAgentRestart, Box: w.box, Ended: w.now().UTC(), LastSeq: last}
	// A run's log lines are at most once and best effort (amadan #325), like
	// the rest of the platform stream; result.json and the audit log are its
	// record. So the ending is logged only by a pass that wrote the run's
	// result itself (read before, read back after) AND whose own done CAS
	// landed. Every pass that still holds a picked snapshot comes back here:
	// the create-only write answers a second time as if it had landed, a
	// result already there is the boot's that ran the command (which logged
	// its own ending), and a stale pass after the run was acknowledged
	// recreates a result but its CAS fails. A lost answer anywhere loses the
	// line rather than risk a second one.
	prior, err := runs.ReadResult(ctx, w.store, rec.Account, rec.App, id)
	if err != nil {
		log.Printf("agent: runs: %s %s: read the result: %v", rec.Host, id, err)
		return
	}
	ours := false
	if prior == nil {
		werr := runs.WriteResult(ctx, w.store, rec.Account, rec.App, id, res)
		got, err := runs.ReadResult(ctx, w.store, rec.Account, rec.App, id)
		if err != nil || got == nil {
			log.Printf("agent: runs: %s %s: result: %v", rec.Host, id, errors.Join(werr, err))
			return
		}
		ours = got.Box == res.Box && got.Ended.Equal(res.Ended) && got.Interrupted == res.Interrupted
	}
	if _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, provision.RunFinish(id)); err == nil {
		if ours {
			w.runEvent(rec, runEndedLine(id, res, true))
		}
	} else if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
		log.Printf("agent: runs: %s %s: done: %v", rec.Host, id, err)
	}
	if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), id); err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, id, err)
	}
	log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id)
}

// armRunsPass arms the runs worker on a box with an activator. Thin, like
// armMovesPass: the behaviour is the worker's, tested there. events must be
// a genuinely nil interface when there is no log store, never a nil
// *applog.Sink (the trap agent.go documents).
func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string, events activator.EventSink) {
	if svc == nil || reg == nil {
		return
	}
	store, err := openDeployStore(context.Background())
	if err != nil {
		log.Printf("agent: runs disabled: %v", err)
		return
	}
	w := newRunsWorker(store, reg.Get, svc, dataDir, BoxLabel(), newBootID())
	w.events = events
	armedRuns.Store(w)
	log.Printf("agent: runs armed (box %s, boot %s)", w.box, w.boot)
}

// withRunsHook gives reconcile the worker's note (run spec §5.2).
func withRunsHook(cfg provision.ReconcileConfig) provision.ReconcileConfig {
	if w := armedRuns.Load(); w != nil && cfg.RunSeen == nil {
		cfg.RunSeen = w.note
	}
	return cfg
}

// runRunsPass acts on what the last reconcile pass noted.
func runRunsPass(ctx context.Context) {
	if w := armedRuns.Load(); w != nil {
		w.pass(ctx)
	}
}

// runRunsBootPass is the worker's one pass at agent start.
func runRunsBootPass(ctx context.Context) {
	if w := armedRuns.Load(); w != nil {
		w.bootPass(ctx)
	}
}