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/platform/auth"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
)

// RearmWait 仅对独立SIGNAL/TIMER换代，固定首次等待的预算；到期事实不能通过新建等待清除。
func (s *WorkflowRecoveryStore) RearmWait(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RearmWaitInput) (recovery.Result, error) {
	return s.mutate(ctx, t, id, key, "workflow.wait.rearm", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
		row, err := workflowRecoveryTarget(g, in.StepDecision)
		if err != nil {
			return nil, err
		}
		node, ok := g.program.Step(row.view.Key)
		if !ok || node.Signal == nil && node.Timer == nil {
			return nil, fault.Invalid("wait rearm requires a SIGNAL or TIMER step")
		}
		if row.view.State != weave.StepWaiting && row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled {
			return nil, fault.Conflict
		}
		if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
			return nil, fault.Conflict
		}
		if err := workflowStepSettled(ctx, tx, g, row); err != nil {
			return nil, err
		}
		origin, err := workflowActivationOrigin(ctx, tx, g, row)
		if err != nil {
			return nil, err
		}
		if origin.sequence >= recovery.MaxWaitActivations {
			return nil, fault.Conflict
		}
		if err := verifyWorkflowWaitHistory(ctx, tx, g, row, origin); err != nil {
			return nil, err
		}
		var w workflowSignalWait
		if node.Signal != nil {
			w, err = readWorkflowSignalWait(ctx, tx, g, row)
			if err != nil {
				return nil, err
			}
			if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil {
				return nil, err
			}
		} else {
			var count int
			if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, t.NamespaceID, id, row.view.ActivationID).Scan(&count); err != nil {
				return nil, err
			}
			if count != 1 {
				return nil, errors.New("timer rearm wait set invalid")
			}
			var kind string
			var name, version *string
			var wake *core.ID
			if err := tx.QueryRowContext(ctx, `SELECT id,kind,state,deadline_at,signal_name,payload_version,correlation_key,winner_signal_id,wake_task_id FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, t.NamespaceID, id, row.view.ActivationID).Scan(&w.id, &kind, &w.state, &w.deadline, &name, &version, &w.correlation, &w.winner, &wake); err != nil {
				return nil, err
			}
			expectedInput, _ := json.Marshal(map[string]uint64{"delay_ms": node.Timer.DelayMS})
			if kind != "TIMER" || name != nil || version != nil || w.correlation != "" || wake != nil || !sameWorkflowJSON(row.input, expectedInput) || !w.deadline.Equal(minimumTime(origin.started.Add(time.Duration(node.Timer.DelayMS)*time.Millisecond), g.view.DeadlineAt)) {
				return nil, errors.New("timer rearm contract mismatch")
			}
		}
		if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) {
			return nil, fault.Conflict
		}
		activation, err := core.NewID()
		if err != nil {
			return nil, err
		}
		wait, err := core.NewID()
		if err != nil {
			return nil, err
		}
		beforeState := row.view.State
		if row.view.State == weave.StepWaiting {
			result, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='CANCELLED',completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='WAITING'`, now, now, t.NamespaceID, in.ActivationID, id, row.view.ID)
			if err := exactlyOne(result, err); err != nil {
				return nil, err
			}
		}
		if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='CANCELLED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, now, now, t.NamespaceID, w.id); err != nil {
			return nil, err
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,?,'WAITING',?,?,?)`, t.NamespaceID, activation, id, row.view.ID, origin.sequence+1, string(origin.input), now, now); err != nil {
			return nil, err
		}
		var name, version *string
		if w.contract != nil {
			name = &w.contract.Name
			version = &w.contract.PayloadVersion
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, wait, id, activation, row.view.Kind, []byte(w.correlation), w.deadline, name, version, now, now); err != nil {
			return nil, err
		}
		result, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='WAITING',current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, activation, now, t.NamespaceID, row.view.ID, row.view.Revision, in.ActivationID)
		if err := exactlyOne(result, err); err != nil {
			return nil, err
		}
		row.view.State, row.view.ActivationID = weave.StepWaiting, &activation
		row.view.ActivationSequence = origin.sequence + 1
		row.view.Revision++
		row.output = nil
		due := w.deadline
		if node.Signal != nil {
			due = now
		}
		if err := saveWorkflowClock(ctx, tx, g, row, string(row.view.Kind), row.view.Revision, due); err != nil {
			return nil, err
		}
		if err := emitEvent(ctx, tx, t.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": id, "state": row.view.State, "activation_id": activation}); err != nil {
			return nil, err
		}
		out.Step = &row.view
		return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": origin.sequence + 1, "previous_state": beforeState, "previous_wait_id": w.id, "previous_wait_state": w.state, "wait_id": wait, "deadline_at": w.deadline, "wait_verified": true}, nil
	})
}

func verifyWorkflowWaitHistory(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, origin workflowActivationBase) error {
	rows, err := tx.QueryContext(ctx, `SELECT id,sequence,state,input,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 33`, g.view.NamespaceID, g.view.ID, row.view.ID)
	if err != nil {
		return err
	}
	defer rows.Close()
	var count uint32
	var last core.ID
	for rows.Next() {
		var id core.ID
		var sequence uint32
		var state string
		var input []byte
		var task *core.ID
		if err := rows.Scan(&id, &sequence, &state, &input, &task); err != nil {
			return err
		}
		count++
		if sequence != count || task != nil || !sameWorkflowJSON(input, origin.input) || id != *row.view.ActivationID && state != "NEEDS_REVIEW" && state != "FAILED" && state != "CANCELLED" {
			return errors.New("wait activation history invalid")
		}
		last = id
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if count != origin.sequence || last != *row.view.ActivationID {
		return errors.New("wait activation history incomplete")
	}
	return nil
}