package mysql

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

	"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/events"
	"git.yuntuo.hk/k2safe/orbit/weave/definition"
	"git.yuntuo.hk/k2safe/orbit/weave/engine"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
)

// workflowSourceError 只用于已核实的来源/契约矛盾；SQL或连接错误不能据此隔离业务结果。
type workflowSourceError string

func (e workflowSourceError) Error() string { return string(e) }

type workflowResult struct {
	event, task, activation core.ID
	revision                uint64
	snapshot                events.TaskResult
	occurred                time.Time
	step                    *workflowStep
}

func readWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, now time.Time) (*workflowResult, error) {
	var state string
	var due time.Time
	var raw []byte
	source := &workflowResult{}
	err := tx.QueryRowContext(ctx, `SELECT event_id,source_task_id,source_activation_id,source_revision,snapshot,state,available_at FROM weave_inbox WHERE namespace_id=? AND id=? AND workflow_id=? FOR UPDATE`, c.ns, c.id, c.workflow).Scan(&source.event, &source.task, &source.activation, &source.revision, &raw, &state, &due)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if state != "PENDING" || due.After(now) {
		return nil, nil
	}
	if _, err := jsonvalue.Canonical(raw); err != nil {
		return nil, workflowSourceError("RESULT_SNAPSHOT_INVALID")
	}
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.DisallowUnknownFields()
	if decoder.Decode(&source.snapshot) != nil || source.snapshot.SchemaVersion != 1 || source.snapshot.TaskID != source.task || source.snapshot.Revision != source.revision || source.revision == 0 || !events.ResultType("task."+strings.ToLower(source.snapshot.State)) {
		return nil, workflowSourceError("RESULT_SNAPSHOT_INVALID")
	}
	var aggregate core.ID
	var kind, eventType string
	var rev uint64
	err = tx.QueryRowContext(ctx, `SELECT aggregate_kind,aggregate_id,aggregate_revision,type,occurred_at FROM orbit_events WHERE namespace_id=? AND id=?`, c.ns, source.event).Scan(&kind, &aggregate, &rev, &eventType, &source.occurred)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, workflowSourceError("RESULT_EVENT_MISSING")
	}
	if err != nil {
		return nil, err
	}
	if kind != "task" || aggregate != source.task || rev != source.revision || eventType != "task."+strings.ToLower(source.snapshot.State) || source.occurred.After(now) {
		return nil, workflowSourceError("RESULT_EVENT_MISMATCH")
	}
	var stepID core.ID
	var direct *core.ID
	err = tx.QueryRowContext(ctx, `SELECT step_id,task_id FROM weave_activations WHERE namespace_id=? AND id=? AND workflow_id=?`, c.ns, source.activation, c.workflow).Scan(&stepID, &direct)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, workflowSourceError("RESULT_ACTIVATION_MISSING")
	}
	if err != nil {
		return nil, err
	}
	for _, step := range g.steps {
		if step.view.ID == stepID {
			source.step = step
			break
		}
	}
	if source.step == nil {
		return nil, workflowSourceError("RESULT_STEP_MISSING")
	}
	node, ok := g.program.Step(source.step.view.Key)
	if !ok || node.Task == nil && node.Poll == nil {
		return nil, workflowSourceError("RESULT_STEP_KIND")
	}
	taskDefinition := node.Task
	if node.Poll != nil {
		taskDefinition = &node.Poll.Task
		var round uint32
		err = tx.QueryRowContext(ctx, `SELECT r.round FROM weave_poll_rounds r JOIN weave_poll_states p ON p.namespace_id=r.namespace_id AND p.workflow_id=r.workflow_id AND p.activation_id=r.activation_id WHERE r.namespace_id=? AND r.workflow_id=? AND r.activation_id=? AND r.task_id=? AND r.round BETWEEN 1 AND p.used_rounds`, c.ns, c.workflow, source.activation, source.task).Scan(&round)
		if errors.Is(err, sql.ErrNoRows) || direct != nil {
			return nil, workflowSourceError("RESULT_POLL_OWNER")
		}
		if err != nil {
			return nil, err
		}
	} else if direct == nil || *direct != source.task {
		return nil, workflowSourceError("RESULT_TASK_OWNER")
	}
	var queue core.ID
	var owner, gate *core.ID
	var ownerKind sql.NullString
	var currentRevision uint64
	err = tx.QueryRowContext(ctx, `SELECT queue_id,owner_kind,owner_id,gate_id,revision FROM pulse_tasks WHERE namespace_id=? AND id=?`, c.ns, source.task).Scan(&queue, &ownerKind, &owner, &gate, &currentRevision)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, workflowSourceError("RESULT_TASK_MISSING")
	}
	if err != nil {
		return nil, err
	}
	if queue != source.snapshot.QueueID || queue != taskDefinition.QueueID || ownerKind.String != engine.TaskOwnerKind || owner == nil || *owner != source.activation || gate == nil || *gate != g.gate || currentRevision < source.revision {
		return nil, workflowSourceError("RESULT_TASK_ASSOCIATION")
	}
	if source.snapshot.CurrentAttemptID != nil {
		var attempt core.ID
		err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_attempts WHERE namespace_id=? AND id=? AND task_id=?`, c.ns, source.snapshot.CurrentAttemptID, source.task).Scan(&attempt)
		if errors.Is(err, sql.ErrNoRows) {
			return nil, workflowSourceError("RESULT_ATTEMPT_ASSOCIATION")
		}
		if err != nil {
			return nil, err
		}
	}
	return source, nil
}

func (r *WorkflowRuntime) consume(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, allow bool, now time.Time) (bool, error) {
	source, err := readWorkflowResult(ctx, tx, g, c, now)
	var invalid workflowSourceError
	if errors.As(err, &invalid) {
		if !g.view.State.Terminal() {
			if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, string(invalid), now); err != nil {
				return false, err
			}
		}
		return true, r.confirmWorkflowResult(ctx, tx, g, c, "QUARANTINED", string(invalid), now)
	}
	if err != nil || source == nil {
		return false, err
	}
	row := source.step
	if row.view.ActivationID == nil || *row.view.ActivationID != source.activation {
		return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", "STALE_ACTIVATION", now)
	}
	var newer int
	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_inbox WHERE namespace_id=? AND source_task_id=? AND source_revision>? AND state='PROCESSED'`, c.ns, source.task, source.revision).Scan(&newer); err != nil {
		return false, err
	}
	if newer > 0 {
		return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", "STALE_RESULT", now)
	}
	if g.view.State.Terminal() || row.view.State != weave.StepActive && row.view.State != weave.StepWaiting {
		return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", "STEP_ALREADY_SETTLED", now)
	}
	state, output, reason := weave.StepSucceeded, source.snapshot.Result, "TASK_RESULT"
	if source.snapshot.EffectUncertain || source.snapshot.State == "NEEDS_REVIEW" {
		state, output, reason = weave.StepNeedsReview, nil, "TASK_RESULT_UNCERTAIN"
	} else if source.snapshot.State == "CANCELLED" {
		state, output, reason = weave.StepCancelled, nil, "TASK_RESULT_CANCELLED"
	} else if source.snapshot.State != "SUCCEEDED" {
		state, output, reason = weave.StepFailed, nil, "TASK_RESULT_FAILED"
	}
	node, _ := g.program.Step(row.view.Key)
	if node.Poll != nil {
		var used, max uint32
		var deadline time.Time
		var current *core.ID
		var resolved *time.Time
		if err := tx.QueryRowContext(ctx, `SELECT used_rounds,max_rounds,deadline_at,current_task_id,resolved_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, c.ns, c.workflow, source.activation).Scan(&used, &max, &deadline, &current, &resolved); err != nil {
			return false, err
		}
		if resolved != nil || current == nil || *current != source.task {
			return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", "STALE_POLL_ROUND", now)
		}
		if node.Poll.FinalSignal != nil {
			handled, _, reason, err := r.driveWorkflowSignal(ctx, tx, g, row, c, source, allow, now)
			if err != nil {
				return false, err
			}
			if handled {
				return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", reason, now)
			}
		}
		// 单轮请求失败不能通过新建业务轮次绕过Pulse预算。
		if g.view.StopKind != nil && (state == weave.StepCancelled || state == weave.StepFailed) {
			reason = "POLL_STOP_REQUEST_SETTLED"
		} else if state != weave.StepSucceeded {
			state, reason = weave.StepNeedsReview, "POLL_REQUEST_REVIEW"
		} else if !source.occurred.Before(deadline) {
			state, output, reason = weave.StepNeedsReview, nil, "POLL_DEADLINE"
		} else {
			outcome, classifyErr := g.program.ClassifyPoll(row.view.Key, output)
			switch {
			case classifyErr != nil:
				state, output, reason = weave.StepNeedsReview, nil, "POLL_RESULT_INVALID"
			case outcome == engine.PollPending:
				state, output, reason = weave.StepWaiting, nil, "POLL_PENDING"
				if g.view.StopKind != nil {
					state, reason = weave.StepCancelled, "POLL_STOP_PENDING_SETTLED"
				} else if used >= max {
					state, reason = weave.StepNeedsReview, "POLL_ROUNDS_EXHAUSTED"
				}
			case outcome == engine.PollFailed:
				state, output, reason = weave.StepFailed, nil, "POLL_BUSINESS_FAILED"
			case outcome == engine.PollNeedsReview:
				state, output, reason = weave.StepNeedsReview, nil, "POLL_BUSINESS_REVIEW"
			}
		}
		var next, settled *time.Time
		if state == weave.StepWaiting {
			v := minimumTime(source.occurred.Add(time.Duration(node.Poll.IntervalMS)*time.Millisecond), deadline)
			next = &v
		} else {
			settled = &now
		}
		if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET current_task_id=NULL,next_poll_at=?,resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, next, settled, now, c.ns, c.workflow, source.activation); err != nil {
			return false, err
		}
		if next != nil {
			if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision+1, *next); err != nil {
				return false, err
			}
		} else if err := deleteWorkflowClock(ctx, tx, g, row); err != nil {
			return false, err
		}
		if settled != nil {
			if node.Poll.FinalSignal != nil {
				w, err := readWorkflowSignalWait(ctx, tx, g, row)
				if err != nil {
					return false, err
				}
				if err := rejectOtherWorkflowSignals(ctx, tx, g, row, w.sequence, "QUERY_SETTLED", now); err != nil {
					return false, err
				}
			}
			waitState := "CANCELLED"
			if state == weave.StepSucceeded {
				waitState = "SATISFIED"
			}
			if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state=?,resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND state='WAITING'`, waitState, now, now, c.ns, c.workflow, source.activation); err != nil {
				return false, err
			}
		}
	} else if state == weave.StepSucceeded && (bytes.Equal(bytes.TrimSpace(output), []byte("null")) || definition.ValidateData(node.Task.ResultType, output) != nil) {
		state, output, reason = weave.StepNeedsReview, nil, "TASK_RESULT_INVALID"
	}
	if err := saveWorkflowResultStep(ctx, tx, g, row, state, output, now); err != nil {
		return false, err
	}
	if state == weave.StepNeedsReview || state == weave.StepFailed || state == weave.StepCancelled {
		next := weave.InstanceNeedsReview
		if (state == weave.StepFailed || state == weave.StepCancelled) && (g.view.StopKind != nil || g.version.Specification.FailurePolicy == "FAIL_AFTER_SETTLED" || g.view.CancelRequestedAt != nil) {
			next = weave.InstanceCancelling
		}
		if err := holdWorkflow(ctx, tx, g, next, reason, now); err != nil {
			return false, err
		}
	} else if err := r.progress(ctx, tx, g, allow, now); err != nil {
		return false, err
	}
	return true, r.confirmWorkflowResult(ctx, tx, g, c, "PROCESSED", reason, now)
}

func saveWorkflowResultStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, state weave.StepState, output json.RawMessage, now time.Time) error {
	var raw any
	if output != nil {
		raw = string(output)
	}
	var completed *time.Time
	if state == weave.StepSucceeded || state == weave.StepFailed || state == weave.StepCancelled {
		completed = &now
	}
	r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state=?`, state, raw, completed, now, g.view.NamespaceID, row.view.ActivationID, g.view.ID, row.view.ID, row.view.State)
	if err := exactlyOne(r, err); err != nil {
		return err
	}
	r, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, state, now, g.view.NamespaceID, row.view.ID, row.view.Revision, row.view.ActivationID)
	if err := exactlyOne(r, err); err != nil {
		return err
	}
	row.view.State = state
	row.view.Revision++
	row.output = output
	r, 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(r, err); err != nil {
		return err
	}
	g.view.Revision++
	return emitEvent(ctx, tx, g.view.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": g.view.ID, "key": row.view.Key, "state": state, "activation_id": row.view.ActivationID})
}

func (r *WorkflowRuntime) confirmWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, state, reason string, now time.Time) error {
	res, err := tx.ExecContext(ctx, `UPDATE weave_inbox SET state=?,processed_at=?,error_code=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND state='PENDING'`, state, now, reason, now, c.ns, c.id, c.workflow)
	if err := exactlyOne(res, err); err != nil {
		return err
	}
	return r.audit(ctx, tx, g, c.id, reason, now)
}