package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/weave/engine"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
)

// WorkflowRuntime 消费持久化结果、根推进与等待时钟检查点；不持有业务执行租约或调用业务系统。
// 每个候选独立短事务，数据库行锁决定唯一推进权，节点内存不保存恢复游标。
type WorkflowRuntime struct {
	db   *sql.DB
	node core.ID
}

func NewWorkflowRuntime(db *sql.DB, node core.ID) (*WorkflowRuntime, error) {
	if db == nil || node.Validate() != nil {
		return nil, errors.New("invalid workflow runtime configuration")
	}
	return &WorkflowRuntime{db: db, node: node}, nil
}

type WorkflowRuntimeStats struct {
	Examined int // Examined 本批取出的持久化候选数，包含被其他节点先处理或跳过的记录。
	Consumed int // Consumed 已提交确认或隔离的内部结果数，事务回滚不计入。
	Advanced int // Advanced 已提交推进的根或等待检查点数，不等于业务任务执行次数。
}

type workflowCandidate struct {
	kind             string
	ns, id, workflow core.ID
	revision         uint64
}

// Tick 同时读取两种到期索引，各取最多limit项再全局选前limit项；不先锁Inbox再反锁父资源。
func (r *WorkflowRuntime) Tick(ctx context.Context, limit int) (WorkflowRuntimeStats, error) {
	var stats WorkflowRuntimeStats
	if limit < 1 || limit > 100 {
		return stats, fault.Invalid("workflow batch limit must be 1 to 100")
	}
	rows, err := r.db.QueryContext(ctx, `(SELECT 'INBOX' kind,namespace_id,id,workflow_id,revision,available_at FROM weave_inbox WHERE state='PENDING' AND available_at<=UTC_TIMESTAMP(6) ORDER BY available_at,namespace_id,id LIMIT ?)
 UNION ALL (SELECT 'ROOT' kind,namespace_id,id,workflow_id,revision,available_at FROM weave_pending_advances WHERE available_at<=UTC_TIMESTAMP(6) ORDER BY available_at,namespace_id,id LIMIT ?)
 ORDER BY available_at,namespace_id,id LIMIT ?`, limit, limit, limit)
	if err != nil {
		return stats, err
	}
	var candidates []workflowCandidate
	for rows.Next() {
		var c workflowCandidate
		var due time.Time
		if err := rows.Scan(&c.kind, &c.ns, &c.id, &c.workflow, &c.revision, &due); err != nil {
			rows.Close()
			return stats, err
		}
		candidates = append(candidates, c)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return stats, err
	}
	var failures []error
	for _, c := range candidates {
		stats.Examined++
		done, err := r.process(ctx, c, true)
		if err != nil {
			// 仅移动同一版本的未消费记录，不能覆盖其他节点刚提交的确认或待续状态。
			// 数据库异常不被归类为业务失败；来源保留供下一批恢复。
			if ctx.Err() == nil {
				err = errors.Join(err, r.deferCandidate(ctx, c))
			}
			failures = append(failures, err)
			if ctx.Err() != nil {
				break
			}
		} else if done {
			if c.kind == "INBOX" {
				stats.Consumed++
			} else {
				stats.Advanced++
			}
		}
	}
	return stats, errors.Join(failures...)
}

func (r *WorkflowRuntime) deferCandidate(ctx context.Context, c workflowCandidate) error {
	q := `UPDATE weave_pending_advances SET available_at=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 30 SECOND),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=? AND revision=?`
	if c.kind == "INBOX" {
		q = `UPDATE weave_inbox SET available_at=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 30 SECOND),error_code='RUNTIME_UNAVAILABLE',revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=? AND revision=? AND state='PENDING'`
	}
	_, err := r.db.ExecContext(ctx, q, c.ns, c.id, c.revision)
	return err
}

func (r *WorkflowRuntime) process(ctx context.Context, c workflowCandidate, skip bool) (bool, error) {
	done := false
	err := WithTx(ctx, r.db, func(tx *sql.Tx) error {
		g, allow, err := lockWorkflowGraph(ctx, tx, c.ns, c.workflow, skip)
		if err != nil || g == nil {
			return err
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		if c.kind == "INBOX" {
			done, err = r.consume(ctx, tx, g, c, allow, now)
			return err
		}
		var step core.ID
		var expected uint64
		var raw []byte
		var due time.Time
		err = tx.QueryRowContext(ctx, `SELECT step_id,expected_step_revision,checkpoint,available_at FROM weave_pending_advances WHERE namespace_id=? AND id=? AND workflow_id=? FOR UPDATE`, c.ns, c.id, c.workflow).Scan(&step, &expected, &raw, &due)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		if due.After(now) {
			return nil
		}
		var checkpoint workflowCheckpoint
		if json.Unmarshal(raw, &checkpoint) != nil || checkpoint.SchemaVersion != 1 {
			return errors.New("workflow checkpoint invalid")
		}
		if checkpoint.Kind == "TIMER" || checkpoint.Kind == "POLL" {
			done, err = r.advanceClock(ctx, tx, g, c, step, expected, checkpoint, allow, now)
			return err
		}
		root := g.steps[workflowRootKey]
		if checkpoint.Kind != "ROOT" || checkpoint.ActivationID != nil || step != root.view.ID || expected > root.view.Revision {
			return errors.New("workflow checkpoint invalid")
		}
		if g.view.State.Terminal() {
			_, err = tx.ExecContext(ctx, `DELETE FROM weave_pending_advances WHERE namespace_id=? AND id=?`, c.ns, c.id)
			return err
		}
		if !allow || g.view.State == weave.InstanceNeedsReview || g.view.State == weave.InstanceCancelling {
			// 暂停/停用仍按原截止计时；核查和失败结清由控制运行器接续，不清除其状态。
			if (g.view.State == weave.InstanceRunning || g.view.State == weave.InstanceWaiting) && !now.Before(g.view.DeadlineAt) {
				if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, "WORKFLOW_DEADLINE", now); err != nil {
					return err
				}
				if err := r.audit(ctx, tx, g, c.id, "WORKFLOW_DEADLINE", now); err != nil {
					return err
				}
			}
			return saveWorkflowAdvance(ctx, tx, g, now.Add(time.Second))
		}
		beforeRevision := g.view.Revision
		if err := r.progress(ctx, tx, g, allow, now); err != nil {
			return err
		}
		if g.view.Revision == beforeRevision { // 额度不足只延后检查点，不逐秒制造业务审计。
			return nil
		}
		if err := r.audit(ctx, tx, g, c.id, "ROOT_ADVANCE", now); err != nil {
			return err
		}
		done = true
		return nil
	})
	return done && err == nil, err
}

func cloneWorkflowGraph(g *workflowGraph) *workflowGraph {
	out := *g
	out.steps = make(map[string]*workflowStep, len(g.steps))
	for key, row := range g.steps {
		copy := *row
		out.steps[key] = &copy
	}
	return &out
}

// progress 在结果已保存后执行有限决策。额度不足只回滚后继决策，不撤销已确认的结果。
func (r *WorkflowRuntime) progress(ctx context.Context, tx *sql.Tx, g *workflowGraph, allow bool, now time.Time) error {
	if g.view.State.Terminal() || g.view.State == weave.InstanceNeedsReview || g.view.State == weave.InstanceCancelling {
		return nil
	}
	allow = allow && now.Before(g.view.DeadlineAt)
	d, err := g.program.Plan(workflowSnapshot(g, allow), 32)
	if err != nil {
		var problem *engine.Problem
		if !errors.As(err, &problem) {
			return err
		}
		return holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, problem.Code, now)
	}
	if !d.Complete && !now.Before(g.view.DeadlineAt) {
		return holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, "WORKFLOW_DEADLINE", now)
	}
	if _, err := tx.ExecContext(ctx, `SAVEPOINT workflow_successors`); err != nil {
		return err
	}
	before := cloneWorkflowGraph(g)
	err = applyWorkflowDecision(ctx, tx, g, d, now)
	if errors.Is(err, fault.QueueFull) || errors.Is(err, fault.ApplicationFull) {
		if _, rollback := tx.ExecContext(ctx, `ROLLBACK TO SAVEPOINT workflow_successors`); rollback != nil {
			return rollback
		}
		*g = *before
		return saveWorkflowAdvance(ctx, tx, g, now.Add(time.Second))
	}
	if err != nil {
		return err
	}
	if !allow && !d.Complete {
		return saveWorkflowAdvance(ctx, tx, g, now.Add(time.Second))
	}
	return nil
}

func holdWorkflow(ctx context.Context, tx *sql.Tx, g *workflowGraph, state weave.InstanceState, reason string, now time.Time) error {
	if g.view.State.Terminal() {
		return errors.New("cannot hold completed workflow")
	}
	if g.view.State == weave.InstanceNeedsReview {
		state = weave.InstanceNeedsReview
	}
	if _, err := tx.ExecContext(ctx, `UPDATE pulse_execution_gates SET paused=TRUE,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, g.view.NamespaceID, g.gate); err != nil {
		return err
	}
	r, err := tx.ExecContext(ctx, `UPDATE weave_instances SET state=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, state, now, g.view.NamespaceID, g.view.ID, g.view.Revision)
	if err := exactlyOne(r, err); err != nil {
		return err
	}
	g.view.State = state
	g.view.Revision++
	return emitEvent(ctx, tx, g.view.NamespaceID, g.view.ID, g.view.Revision, "workflow", "workflow.blocked", map[string]any{"state": state, "reason": reason})
}

func (r *WorkflowRuntime) audit(ctx context.Context, tx *sql.Tx, g *workflowGraph, operation core.ID, reason string, now time.Time) error {
	id, err := core.NewID()
	if err != nil {
		return err
	}
	// 迟到证据也占用新的实例历史版本，避免与同一实例版本先前的推进事件冲突。
	res, err := tx.ExecContext(ctx, `UPDATE weave_instances SET revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, now, g.view.NamespaceID, g.view.ID, g.view.Revision)
	if err := exactlyOne(res, err); err != nil {
		return err
	}
	g.view.Revision++
	meta := map[string]any{"state": g.view.State, "revision": g.view.Revision, "node_id": r.node, "operation_id": operation, "reason": reason}
	raw, err := json.Marshal(meta)
	if err != nil {
		return err
	}
	if _, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`before`,`after`) VALUES (?,?,?,?,'workflow.advance','workflow',?,?,'null',?)", g.view.NamespaceID, id, id, "system:weave:"+string(r.node), g.view.ID, reason, string(raw)); err != nil {
		return err
	}
	return emitEvent(ctx, tx, g.view.NamespaceID, g.view.ID, g.view.Revision, "workflow", "workflow.advanced", meta)
}