package mill

import (
	"os"
	"sync"

	"tangled.org/core/spindle/models"

	millv1 "tangled.org/core/spindle/mill/proto/gen"
)

// leaseState is the mill's view of a remote attempt. It collapses the doc's
// full state diagram to what the mill actually needs to drive cleanup and the
// commit-and-wait in RunStep.
type leaseState int32

const (
	// won a bid, executor is holding a seat, not yet committed.
	leaseReserved leaseState = iota
	// CommitLease has been sent; the executor may already be running the job,
	// but the mill has not necessarily received Committed yet.
	leaseCommitting
	// CommitLease sent and acked; the job is running on the executor.
	leaseRunning
	// a terminal result arrived (or we gave up). no further action.
	leaseDone
)

type cancelAction int

const (
	cancelNoop cancelAction = iota
	cancelLocal
	cancelRemote
)

// RemoteLease is the mill-side fencing token for one placed job.
type RemoteLease struct {
	id     string
	nodeID string
	engine string
	wid    models.WorkflowId // the job this lease carries; set once placed
	// restored from persistence after a mill restart: no RunStep waits on it,
	// so terminals and death are authored directly. Set before publication,
	// never mutated.
	orphaned bool

	mu       sync.Mutex
	state    leaseState
	cancel   bool
	reason   string
	released bool
	terminal chan *millv1.AttemptResult // buffered(1); RunStep waits here
	dead     chan struct{}              // closed when the executor is lost past grace
	deadOnce sync.Once

	finishMu    sync.Mutex
	cleanupOnce sync.Once

	// log file for relayed lines, opened lazily and held open for the lease's
	// life so we don't reopen per line. Guarded by logMu, closed on cleanup.
	logMu     sync.Mutex
	logFile   *os.File
	logClosed bool
}

func newLease(id, nodeID, engine string) *RemoteLease {
	return &RemoteLease{
		id:       id,
		nodeID:   nodeID,
		engine:   engine,
		state:    leaseReserved,
		terminal: make(chan *millv1.AttemptResult, 1),
		dead:     make(chan struct{}),
	}
}

func (l *RemoteLease) setState(s leaseState) {
	l.mu.Lock()
	l.state = s
	l.mu.Unlock()
}

func (l *RemoteLease) markCommitting() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.state == leaseDone {
		return false
	}
	if l.state == leaseReserved {
		l.state = leaseCommitting
	}
	return true
}

func (l *RemoteLease) markRunning() {
	l.mu.Lock()
	if l.state != leaseDone {
		l.state = leaseRunning
	}
	l.mu.Unlock()
}

func (l *RemoteLease) getState() leaseState {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.state
}

// markDone CASes to done; exactly one caller wins.
func (l *RemoteLease) markDone() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.state == leaseDone {
		return false
	}
	l.state = leaseDone
	return true
}

func (l *RemoteLease) requestCancel(reason string) cancelAction {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.state == leaseDone {
		return cancelNoop
	}
	l.cancel = true
	l.reason = reason
	if l.state == leaseReserved {
		return cancelLocal
	}
	return cancelRemote
}

func (l *RemoteLease) cancelRequested() (bool, string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.cancel, l.reason
}

func (l *RemoteLease) releaseState() (leaseState, bool) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.released = true
	return l.state, l.cancel
}

func (l *RemoteLease) cleanupReady() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.released && l.state == leaseDone
}

func (l *RemoteLease) deliverCancelled(reason string) {
	l.deliverTerminal(&millv1.AttemptResult{
		LeaseId:        l.id,
		TerminalStatus: string(models.StatusKindCancelled),
		Error:          reason,
	})
}

// deliverTerminal hands the terminal result to a waiting RunStep without
// blocking. Idempotent: a duplicate (e.g. replayed on reconnect) is dropped
// because the channel is buffered to one and the lease is already done.
func (l *RemoteLease) deliverTerminal(res *millv1.AttemptResult) {
	l.mu.Lock()
	if l.state == leaseDone {
		l.mu.Unlock()
		return
	}
	l.state = leaseDone
	l.mu.Unlock()

	select {
	case l.terminal <- res:
	default:
	}
}

// markDead signals that the executor running this lease is gone for good.
func (l *RemoteLease) markDead() {
	l.deadOnce.Do(func() { close(l.dead) })
}

// writeLog appends a relayed log line to the lease's log file, opening it once
// and holding the handle open for the lease's life (relayed lines for a lease
// all arrive on one session reader goroutine, so this is serialized). A write
// after closeLog is dropped so a replayed line can't resurrect the file.
func (l *RemoteLease) writeLog(path string, line []byte) error {
	l.logMu.Lock()
	defer l.logMu.Unlock()
	if l.logClosed {
		return nil
	}
	if l.logFile == nil {
		f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
		if err != nil {
			return err
		}
		l.logFile = f
	}
	_, err := l.logFile.Write(line)
	return err
}

// Idempotent.
func (l *RemoteLease) closeLog() {
	l.logMu.Lock()
	defer l.logMu.Unlock()
	l.logClosed = true
	if l.logFile != nil {
		_ = l.logFile.Close()
		l.logFile = nil
	}
}

// remoteStep is the mill's synthetic single step. The real steps run on the
// executor; the mill never mirrors them.
type remoteStep struct{}

func (remoteStep) Name() string          { return "remote execution" }
func (remoteStep) Command() string       { return "" }
func (remoteStep) Kind() models.StepKind { return models.StepKindSystem }

// millSlot is what AcquireWorkflowSlot returns; its Release unwinds placement.
type millSlot struct {
	fleet *Mill
	lease *RemoteLease
	once  sync.Once
}

func (s *millSlot) Release() {
	if s == nil {
		return
	}
	s.once.Do(func() { s.fleet.releaseSlot(s) })
}
