runner.go

package legatus

import (

	"context"

	"io"

	"log/slog"

	"sync"

	"time"

	"go.bigb.es/auxilia/culpa"

	"go.bigb.es/auxilia/scribe"

)

// Runner executes flows across an inventory. It pools one Conn per host and

// carries a SharedStore so sequential Run calls can pass data between groups

// (e.g. exits → relay). The Runner reaches hosts only through the Dialer/Conn

// interfaces, never SSH directly (IV1, PC1).

type Runner struct {

	inv         *Inventory

	dialer      Dialer

	concurrency int

	policy      ErrorPolicy

	reporter    Reporter

	prompter    Prompter

	logger      *slog.Logger

	dryRun      bool

	out         io.Writer

	shared *SharedStore

	mu   sync.Mutex

	pool map[string]Conn

}

// NewRunner builds a Runner. Defaults: concurrency 1, ContinueOnError, no-op

// reporter, non-interactive prompter, slog.Default logger.

func NewRunner(inv *Inventory, opts ...Option) *Runner {

	r := &Runner{

		inv:         inv,

		concurrency: 1,

		policy:      ContinueOnError,

		reporter:    NopReporter{},

		prompter:    DenyPrompter{},

		logger:      slog.Default(),

		out:         io.Discard,

		shared:      NewSharedStore(),

		pool:        map[string]Conn{},

	}

	for _, o := range opts {

		o(r)

	}

	return r

}

// Shared returns the cross-host store, persistent across Run calls.

func (r *Runner) Shared() *SharedStore { return r.shared }

// Run executes flow against the hosts matching sel (all hosts if none), fanning

// out across an ordered worker pool capped at the configured concurrency. It

// always returns a result per host. Under FailFast, once a host fails the run

// context is cancelled and hosts not yet dispatched are recorded as cancelled

// without dialing — because jobs are fed in inventory order, with concurrency 1

// this cancellation is deterministic.

func (r *Runner) Run(ctx context.Context, flow Flow, sel ...Selector) RunResult {

	hosts := r.inv.Hosts(sel...)

	results := make([]FlowResult, len(hosts))

	conc := r.concurrency

	if conc < 1 {

		conc = 1

	}

	if conc > len(hosts) {

		conc = len(hosts)

	}

	runCtx, cancel := context.WithCancel(ctx)

	defer cancel()

	type job struct {

		idx int

		h   *Host

	}

	jobs := make(chan job)

	var wg sync.WaitGroup

	var failOnce sync.Once

	for w := 0; w < conc; w++ {

		wg.Add(1)

		go func() {

			defer wg.Done()

			for j := range jobs {

				fr := r.runHost(runCtx, flow, j.h)

				results[j.idx] = fr

				if r.policy == FailFast && fr.Status == StatusFailed {

					failOnce.Do(cancel)

				}

			}

		}()

	}

	for i, h := range hosts {

		if r.policy == FailFast {

			select {

			case <-runCtx.Done():

				results[i] = r.cancelledFlow(flow, h, runCtx.Err())

				continue

			default:

			}

		}

		jobs <- job{idx: i, h: h}

	}

	close(jobs)

	wg.Wait()

	rr := RunResult{Flows: results}

	r.reporter.RunDone(rr)

	return rr

}

func (r *Runner) cancelledFlow(flow Flow, h *Host, err error) FlowResult {

	sr := StepResult{Step: "cancelled", Status: StatusFailed, Err: err, Started: time.Now()}

	return FlowResult{Flow: flow.Name, Host: h.Name, Steps: []StepResult{sr}, Status: StatusFailed, Params: map[string]any{}}

}

func (r *Runner) runHost(ctx context.Context, flow Flow, h *Host) FlowResult {

	hlog := r.logger.With("host", h.Name, "flow", flow.Name)

	r.reporter.FlowStart(h.Name, flow.Name)

	fr := FlowResult{Flow: flow.Name, Host: h.Name}

	conn, err := r.connFor(ctx, h)

	if err != nil {

		err = culpa.Wrap(err, "dial host "+h.Name)

		sr := StepResult{Step: "dial", Status: StatusFailed, Err: err, Started: time.Now()}

		hlog.Error("dial failed", scribe.Err(err))

		r.reporter.StepDone(h.Name, sr)

		fr.Steps = append(fr.Steps, sr)

		fr.Status = StatusFailed

		fr.Params = map[string]any{}

		r.reporter.FlowDone(h.Name, fr)

		return fr

	}

	fc := NewFlowCtx(ctx, h, conn,

		WithFlowLogger(hlog),

		WithFlowDryRun(r.dryRun),

		WithFlowPrompter(r.prompter),

		WithFlowShared(r.shared),

		WithFlowOutput(r.out),

	)

	for _, step := range flow.Steps {

		sr := r.runStep(fc, hlog, h.Name, step)

		// Record change before the next step's Check so later steps can gate on it

		// (steps.When/OnChanged); WouldChange counts so gates fire in dry-run too.

		fc.setChanged(sr.Step, sr.Status == StatusChanged || sr.Status == StatusWouldChange)

		fr.Steps = append(fr.Steps, sr)

		if sr.Status == StatusFailed {

			break // fail-fast within a flow (IV7)

		}

	}

	fr.Status = aggregate(fr.Steps)

	fr.Params = fc.snapshot()

	r.reporter.FlowDone(h.Name, fr)

	return fr

}

// runStep runs one step's check/apply lifecycle, emitting structured logs and

// reporter events with no step-author involvement (PC5).

func (r *Runner) runStep(fc *FlowCtx, hlog *slog.Logger, host string, step Step) StepResult {

	name := step.Name()

	slogger := hlog.With("step", name)

	fc.log = slogger // per-step tag; params and other fields persist across steps

	r.reporter.StepStart(host, name)

	started := time.Now()

	slogger.Debug("step starting")

	sr := StepResult{Step: name, Started: started}

	need, err := step.Check(fc)

	if err != nil {

		sr.Status = StatusFailed

		sr.Err = culpa.Wrap(err, "check "+name)

		sr.Duration = time.Since(started)

		slogger.Error("step failed", "stage", "check", scribe.Err(sr.Err))

		r.reporter.StepDone(host, sr)

		return sr

	}

	slogger.Debug("check complete", "need", need)

	if !need {

		sr.Status = StatusSkipped

		sr.Duration = time.Since(started)

		slogger.Info("step skipped")

		r.reporter.StepDone(host, sr)

		return sr

	}

	if fc.dryRun {

		sr.Status = StatusWouldChange

		sr.Duration = time.Since(started)

		slogger.Info("step would change (dry-run)")

		r.reporter.StepDone(host, sr)

		return sr

	}

	out, err := step.Apply(fc)

	sr.Duration = time.Since(started)

	if err != nil {

		sr.Status = StatusFailed

		sr.Err = culpa.Wrap(err, "apply "+name)

		slogger.Error("step failed", "stage", "apply", scribe.Err(sr.Err))

		r.reporter.StepDone(host, sr)

		return sr

	}

	sr.Outcome = out

	if out.Changed {

		sr.Status = StatusChanged

	} else {

		sr.Status = StatusOK

	}

	slogger.Info("step applied", "status", sr.Status.String(), "changed", out.Changed, "dur", sr.Duration)

	r.reporter.StepDone(host, sr)

	return sr

}

func (r *Runner) connFor(ctx context.Context, h *Host) (Conn, error) {

	r.mu.Lock()

	c, ok := r.pool[h.Name]

	r.mu.Unlock()

	if ok {

		return c, nil

	}

	if r.dialer == nil {

		return nil, culpa.WithCode(culpa.New("legatus: no dialer configured"), "LEGATUS_NO_DIALER")

	}

	c, err := r.dialer.Dial(ctx, h)

	if err != nil {

		return nil, err

	}

	r.mu.Lock()

	if existing, ok := r.pool[h.Name]; ok { // lost a race; keep the first

		r.mu.Unlock()

		_ = c.Close()

		return existing, nil

	}

	r.pool[h.Name] = c

	r.mu.Unlock()

	return c, nil

}

// Close closes all pooled connections. It also satisfies steward's Stopper.

func (r *Runner) Close() error {

	r.mu.Lock()

	defer r.mu.Unlock()

	var firstErr error

	for name, c := range r.pool {

		if err := c.Close(); err != nil && firstErr == nil {

			firstErr = culpa.Wrap(err, "close conn "+name)

		}

		delete(r.pool, name)

	}

	return firstErr

}

// Stop closes pooled connections (steward Stopper).

func (r *Runner) Stop(context.Context) error { return r.Close() }