package executor

import (
	"context"
	"io"
	"sync"
	"time"

	"github.com/hpcloud/tail"

	"tangled.org/core/api/tangled"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"

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

// observeLoop watches the executor's own eventstream and relays the status rows
// belonging to active leases. This is the "observe-by-cursor" relay: the jobs
// write status to the real db/notifier exactly as a standalone spindle would,
// and we forward what is relevant without touching the execution path.
func (e *Executor) observeLoop(ctx context.Context) {
	sub := e.n.Subscribe()
	defer e.n.Unsubscribe(sub)

	// start past existing history; only relay rows produced from now on.
	cursor := time.Now().UnixNano()

	drain := func() {
		for {
			evs, err := e.db.GetEvents(cursor, 100)
			if err != nil {
				e.l.Error("observe GetEvents failed", "err", err)
				return
			}
			for _, ev := range evs {
				cursor = ev.Created
				if ev.Nsid != tangled.PipelineStatusNSID {
					continue
				}
				st, ok := parseStatus(ev.EventJson)
				if !ok {
					continue
				}
				e.onStatusRow(st)
			}
			if len(evs) < 100 {
				return
			}
		}
	}

	// safety tick in case a notify is ever missed. The notifier is the primary
	// wakeup; this only backstops a dropped signal, so it can be coarse.
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-sub:
		case <-ticker.C:
		}
		drain()
	}
}

func (e *Executor) onStatusRow(st *tangled.PipelineStatus) {
	res := e.reservationFor(st.Pipeline, st.Workflow)
	if res == nil {
		return
	}

	kind := models.StatusKind(st.Status)
	switch {
	case kind.IsFinish():
		e.finishJob(res, st)
	case kind == models.StatusKindRunning:
		e.relayMu.Lock()
		e.send(e.relay.appendStatus(res.leaseID, st, st.Pipeline))
		e.relayMu.Unlock()
	}
}

// finishJob flushes the log tail, then relays the terminal as an AttemptResult
// (which the mill turns into the terminal status row), then forgets the lease.
// Tail flush happens first so all log offsets precede the terminal offset.
func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) {
	cleanup, cancelled, ok := e.finishReservation(res)
	if !ok {
		return
	}
	cleanup()

	e.relayMu.Lock()
	terminalStatus := st.Status
	if cancelled {
		terminalStatus = string(models.StatusKindCancelled)
	}
	e.send(e.relay.appendTerminal(res.leaseID, terminalStatus, st))
	e.relayMu.Unlock()

	e.pushSnapshot()
}

// reservationFor finds the active reservation whose workflow matches a status
// row, by pipeline aturi + workflow name. Active reservations are few (bounded
// by seats), so a scan is cheaper than maintaining a parallel index.
func (e *Executor) reservationFor(pipelineAturi, workflow string) *reservation {
	e.mu.Lock()
	defer e.mu.Unlock()
	for _, res := range e.active {
		if string(res.wid.PipelineId.AtUri()) == pipelineAturi && res.wid.Name == workflow {
			return res
		}
	}
	return nil
}

func (e *Executor) relayLogLine(leaseID, text string) {
	e.relayMu.Lock()
	defer e.relayMu.Unlock()
	e.send(e.relay.appendLog(leaseID, []byte(text)))
}

// startTail follows the job's local log file and relays each line. The lines are
// already-encoded models.LogLine JSON, so we forward them verbatim.
func (e *Executor) startTail(res *reservation) {
	path := models.LogFilePath(e.cfg.Server.LogDir, res.wid)
	t, err := tail.TailFile(path, tail.Config{
		Follow:    true,
		ReOpen:    true,
		MustExist: false,
		Location:  &tail.SeekInfo{Offset: 0, Whence: io.SeekStart},
		Logger:    tail.DiscardingLogger,
	})
	if err != nil {
		e.l.Error("tail log file failed", "wid", res.wid, "err", err)
		return
	}

	done := make(chan struct{})
	go func() {
		defer close(done)
		for line := range t.Lines {
			if line == nil || line.Err != nil {
				continue
			}
			e.relayLogLine(res.leaseID, line.Text)
		}
	}()

	var once sync.Once
	res.stopTail = func() {
		once.Do(func() {
			// Stop() blocks until tailing ends and t.Lines is closed, so the
			// consumer goroutine then drains the remaining (finite) buffered
			// lines and closes done. Waiting on done in full guarantees every
			// log line is relayed before finishJob sends the terminal — without
			// it a slow drain lets the terminal overtake a log line, which the
			// mill then drops against its sealed log file.
			_ = t.Stop()
			<-done
		})
	}
}

// memVault is a temporary in-memory secrets manager holding the secrets the
// mill handed over at commit. They never touch disk on the executor.
type memVault struct {
	secrets []secrets.UnlockedSecret
}

func newMemVault(pb []*millv1.Secret) *memVault {
	s := make([]secrets.UnlockedSecret, len(pb))
	for i, x := range pb {
		s[i] = secrets.UnlockedSecret{Key: x.GetKey(), Value: x.GetValue()}
	}
	return &memVault{secrets: s}
}

func (v *memVault) GetSecretsUnlocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) {
	return v.secrets, nil
}
func (v *memVault) GetSecretsLocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.LockedSecret, error) {
	return nil, nil
}
func (v *memVault) AddSecret(ctx context.Context, s secrets.UnlockedSecret) error { return nil }
func (v *memVault) RemoveSecret(ctx context.Context, s secrets.Secret[any]) error { return nil }

var _ secrets.Manager = (*memVault)(nil)
