package mill

import (
	"time"

	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
)

// mill_leases.state values. Committing persists as reserved: a crash mid-commit
// is indistinguishable from one before it, and reconciliation treats both the
// same (the executor either holds the lease or it doesn't).
const (
	leaseRowReserved = "reserved"
	leaseRowRunning  = "running"
)

func (m *Mill) persistLease(lease *RemoteLease, state string) error {
	if m.db == nil {
		return nil
	}
	return m.db.SaveMillLease(db.MillLease{
		LeaseID:  lease.id,
		NodeID:   lease.nodeID,
		Engine:   lease.engine,
		Knot:     lease.wid.Knot,
		Rkey:     lease.wid.Rkey,
		Workflow: lease.wid.Name,
		State:    state,
	})
}

func (m *Mill) deleteLeaseRow(leaseID string) {
	if m.db == nil {
		return
	}
	if err := m.db.DeleteMillLease(leaseID); err != nil {
		m.l.Error("delete mill lease row", "lease", leaseID, "err", err)
	}
}

// RestoreState rebuilds leases and relay cursors persisted by a previous run.
// Restored leases are orphaned: no RunStep waits on them, so their terminal
// (or their executor's death) is authored directly as a status row.
func (m *Mill) RestoreState() error {
	if m.db == nil {
		return nil
	}
	cursors, err := m.db.ListExecutorCursors()
	if err != nil {
		return err
	}
	rows, err := m.db.ListMillLeases()
	if err != nil {
		return err
	}

	m.mu.Lock()
	for node, offset := range cursors {
		m.nodeOffset[node] = offset
	}
	for _, r := range rows {
		lease := newLease(r.LeaseID, r.NodeID, r.Engine)
		lease.wid = models.WorkflowId{
			PipelineId: models.PipelineId{Knot: r.Knot, Rkey: r.Rkey},
			Name:       r.Workflow,
		}
		lease.orphaned = true
		if r.State == leaseRowRunning {
			lease.state = leaseRunning
		}
		m.leases[r.LeaseID] = lease
	}
	restored := len(rows)
	m.mu.Unlock()

	if restored > 0 {
		m.l.Info("restored mill leases from previous run", "leases", restored, "cursors", len(cursors))
		time.AfterFunc(m.cfg.ReconnectGrace, m.sweepUnclaimedOrphans)
	}
	return nil
}

func (m *Mill) sweepUnclaimedOrphans() {
	m.mu.Lock()
	var unclaimed []*RemoteLease
	for _, lease := range m.leases {
		if !lease.orphaned {
			continue
		}
		if sess := m.sessions[lease.nodeID]; sess == nil || sess.disconnected {
			unclaimed = append(unclaimed, lease)
		}
	}
	m.mu.Unlock()

	reason := "executor did not reconnect after mill restart"
	for _, lease := range unclaimed {
		m.l.Warn("failing unclaimed restored lease", "lease", lease.id, "node", lease.nodeID)
		if err := m.finishOrphan(lease, string(models.StatusKindFailed), &reason, nil); err != nil {
			m.l.Error("finish unclaimed restored lease", "lease", lease.id, "err", err)
		}
	}
}

// reconcileOrphans fails restored leases the executor no longer holds. Only
// orphaned leases are touched, so an in-flight bid can never race this.
func (m *Mill) reconcileOrphans(nodeID string, activeLeaseIDs []string) error {
	active := make(map[string]struct{}, len(activeLeaseIDs))
	for _, id := range activeLeaseIDs {
		active[id] = struct{}{}
	}

	m.mu.Lock()
	var gone []*RemoteLease
	for _, lease := range m.leases {
		if lease.orphaned && lease.nodeID == nodeID {
			if _, ok := active[lease.id]; !ok {
				gone = append(gone, lease)
			}
		}
	}
	m.mu.Unlock()

	reason := "executor no longer holds lease after mill restart"
	for _, lease := range gone {
		m.l.Warn("failing restored lease dropped by executor", "lease", lease.id, "node", nodeID)
		if err := m.finishOrphan(lease, string(models.StatusKindFailed), &reason, nil); err != nil {
			return err
		}
	}
	return nil
}

// finishOrphan authors the terminal status row (there is no RunStep waiter to
// do it) and unwinds the lease. markDone makes duplicates no-ops.
func (m *Mill) finishOrphan(lease *RemoteLease, status string, errMsg *string, exitCode *int64) error {
	lease.finishMu.Lock()
	defer lease.finishMu.Unlock()
	if lease.getState() == leaseDone {
		return nil
	}
	if m.db != nil {
		if err := m.db.CompleteOrphanMillLease(
			lease.id,
			string(lease.wid.PipelineId.AtUri()),
			lease.wid.Name,
			status,
			errMsg,
			exitCode,
			m.n,
		); err != nil {
			return err
		}
	}
	lease.markDone()
	m.cleanupLease(lease)
	return nil
}
