package mysql

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

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/weave/definition"
	"git.yuntuo.hk/k2safe/orbit/weave/engine"
	"git.yuntuo.hk/k2safe/orbit/weave/instance"
)

// WorkflowInstanceStore 以应用额度锁串行受理，在一个事务内建立全部归属及初始工作。
// 公共HTTP与内部运行器复用同一套固定版本、幂等与事务规则。
type WorkflowInstanceStore struct{ *TaskStore }

func NewWorkflowInstanceStore(db *sql.DB) *WorkflowInstanceStore {
	return &WorkflowInstanceStore{NewTaskStore(db)}
}

var _ instance.Store = (*WorkflowInstanceStore)(nil)

// 根步骤是实例级有限推进的检查点；$不属于合法的用户步骤字符，不能与定义冲突。
const workflowRootKey = "$root"

type workflowGraph struct {
	view    instance.View
	gate    core.ID
	version definition.VersionView
	program *engine.Program
	steps   map[string]*workflowStep
}
type workflowStep struct {
	view   instance.StepView
	input  json.RawMessage
	output json.RawMessage
}
type workflowPolicy struct {
	SchemaVersion   uint32  `json:"schema_version"`    // SchemaVersion 冻结策略快照的结构版本，当前为1。
	VersionID       core.ID `json:"version_id"`        // VersionID 固定的不可变定义版本，读取时核对实例归属。
	ContentDigest   string  `json:"content_digest"`    // ContentDigest 固定定义的SHA-256摘要，防止策略和规则错配。
	FailurePolicy   string  `json:"failure_policy"`    // FailurePolicy 原定义的失败结清策略，不随新发布版本变化。
	MaxTaskAttempts uint32  `json:"max_task_attempts"` // MaxTaskAttempts 所有路径的静态尝试上界，实际预算由任务与轮次独立保存。
}

func readWorkflowInstance(ctx context.Context, tx *sql.Tx, ns, id core.ID, lock bool) (instance.View, core.ID, error) {
	var v instance.View
	var gate core.ID
	q := `SELECT i.id,i.namespace_id,i.definition_id,i.definition_version_id,v.version,i.gate_id,i.state,i.revision,i.paused,i.control_revision,i.input,i.output,COALESCE(i.business_key,''),i.deadline_at,i.cancel_requested_at,i.stop_kind,i.stop_requested_at,i.completed_at,i.created_at,i.updated_at
 FROM weave_instances i JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.definition_id=i.definition_id AND v.id=i.definition_version_id
 JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id
 JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id
 WHERE i.namespace_id=? AND i.id=?`
	if lock {
		q += ` FOR SHARE`
	}
	var input, output []byte
	err := tx.QueryRowContext(ctx, q, ns, id).Scan(&v.ID, &v.NamespaceID, &v.DefinitionID, &v.DefinitionVersionID, &v.Version, &gate, &v.State, &v.Revision, &v.Paused, &v.ControlRevision, &input, &output, &v.BusinessKey, &v.DeadlineAt, &v.CancelRequestedAt, &v.StopKind, &v.StopRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt)
	v.Input, v.Output = input, output
	if errors.Is(err, sql.ErrNoRows) {
		err = fault.NotFound
	}
	return v, gate, err
}
func instanceVersion(ctx context.Context, tx *sql.Tx, v instance.View) (definition.VersionView, error) {
	ver, err := readWorkflowVersion(ctx, tx, v.NamespaceID, v.DefinitionID, v.Version)
	if err != nil {
		return ver, err
	}
	if ver.ID != v.DefinitionVersionID {
		return ver, errors.New("workflow fixed version mismatch")
	}
	if definition.ValidateData(ver.Specification.InputType, v.Input) != nil {
		return ver, errors.New("workflow stored input invalid")
	}
	if v.State == "SUCCEEDED" && definition.ValidateData(ver.Specification.Body.OutputType, v.Output) != nil {
		return ver, errors.New("workflow stored output invalid")
	}
	var raw []byte
	if err := tx.QueryRowContext(ctx, `SELECT policy_snapshot FROM weave_instances WHERE namespace_id=? AND id=?`, v.NamespaceID, v.ID).Scan(&raw); err != nil {
		return ver, err
	}
	var policy workflowPolicy
	if json.Unmarshal(raw, &policy) != nil || policy != (workflowPolicy{1, ver.ID, ver.ContentDigest, ver.Specification.FailurePolicy, ver.Analysis.MaxTaskAttempts}) {
		return ver, errors.New("workflow frozen policy mismatch")
	}
	return ver, nil
}
func lockWorkflowQueues(ctx context.Context, tx *sql.Tx, ns core.ID, p auth.Permissions, queues []core.ID) error {
	// Checked.Queues始终排序去重。这里直接取排他锁，禁止先共享后升级。
	for _, id := range queues {
		if !p.AllowsQueue(id) {
			return fault.NotFound
		}
		if _, _, err := lockQueueAfterApplication(ctx, tx, ns, id); err != nil {
			return err
		}
	}
	return nil
}

func (s *WorkflowInstanceStore) Start(ctx context.Context, t auth.Token, key string, in instance.StartInput) (instance.StartResult, error) {
	var out instance.StartResult
	err := s.transaction(ctx, t, instance.StartAction, func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
			return err
		}
		id, err := core.NewID()
		if err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "workflow.start", "", key, "workflow", id, in)
		if err != nil {
			return err
		}
		if _, err := readWorkflowDefinition(ctx, tx, t.NamespaceID, in.DefinitionID, false); err != nil {
			return err
		}
		version, err := readWorkflowVersion(ctx, tx, t.NamespaceID, in.DefinitionID, in.Version)
		if err != nil {
			return err
		}
		if len(version.Analysis.Queues) > 0 && !p.Allows("tasks:submit") {
			return fault.Forbidden
		}
		if err := lockWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil {
			return err
		}
		if cached != nil {
			if err := json.Unmarshal(cached, &out); err != nil {
				return err
			}
			current, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, out.Instance.ID, false)
			if err != nil {
				return err
			}
			if current.DefinitionVersionID != version.ID || out.Instance.DefinitionVersionID != version.ID || out.Instance.NamespaceID != t.NamespaceID || !current.CreatedAt.Equal(out.Instance.CreatedAt) {
				return errors.New("workflow start receipt mismatch")
			}
			out.Deduplicated = true
			return nil
		}
		if err := definition.ValidateData(version.Specification.InputType, in.Input); err != nil {
			return fault.Invalid("workflow input does not match fixed version")
		}
		var count int
		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM (SELECT id FROM weave_instances WHERE namespace_id=? AND state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') LIMIT ?) active_workflows`, t.NamespaceID, instance.MaxUnfinished).Scan(&count); err != nil {
			return err
		}
		if count >= instance.MaxUnfinished {
			return fault.New("WORKFLOW_INSTANCE_LIMIT", "namespace unfinished workflow limit reached")
		}
		program, err := engine.Compile(version.Specification)
		if err != nil {
			return err
		}
		decision, err := program.Plan(engine.Snapshot{WorkflowID: id, BusinessKey: in.BusinessKey, Input: in.Input, Steps: map[string]engine.State{}, AllowNew: true}, 32)
		if err != nil {
			return fault.Invalid("workflow initial step input is invalid")
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		gate, err := core.NewID()
		if err != nil {
			return err
		}
		deadline := now.Add(time.Duration(version.Specification.TimeoutMS) * time.Millisecond)
		policy, err := jsonwire.Marshal(workflowPolicy{1, version.ID, version.ContentDigest, version.Specification.FailurePolicy, version.Analysis.MaxTaskAttempts})
		if err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO pulse_execution_gates(namespace_id,id,created_at,updated_at) VALUES (?,?,?,?)`, t.NamespaceID, gate, now, now); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,business_key,deadline_at,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, in.DefinitionID, version.ID, gate, string(in.Input), string(policy), businessKeyValue(in.BusinessKey), deadline, now, now); err != nil {
			return err
		}
		graph := workflowGraph{gate: gate, version: version, program: program, steps: map[string]*workflowStep{}}
		graph.view, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
		if err != nil {
			return err
		}
		if err := createWorkflowSteps(ctx, tx, &graph, now); err != nil {
			return err
		}
		if err := applyWorkflowDecision(ctx, tx, &graph, decision, now); err != nil {
			return err
		}
		out.Instance, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
		if err != nil {
			return err
		}
		meta := map[string]any{"definition_id": in.DefinitionID, "definition_version_id": version.ID, "state": out.Instance.State, "deadline_at": deadline, "steps": version.Analysis.Steps}
		if err := emitEvent(ctx, tx, t.NamespaceID, id, out.Instance.Revision, "workflow", "workflow.started", meta); err != nil {
			return err
		}
		if err := appendAudit(ctx, tx, t, identity, id, "workflow", "workflow.start", "start fixed workflow version", nil, meta); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}

func (s *WorkflowInstanceStore) Get(ctx context.Context, t auth.Token, id core.ID) (instance.View, error) {
	var out instance.View
	err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
		var err error
		out, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
		if err != nil {
			return err
		}
		version, err := instanceVersion(ctx, tx, out)
		if err != nil {
			return err
		}
		return authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues)
	})
	return out, err
}

// readWorkflowForQuery 固定定义授权全部队列后再锁实例，保持与运行器相同的父子锁顺序。
func readWorkflowForQuery(ctx context.Context, tx *sql.Tx, ns, id core.ID, p auth.Permissions) (instance.View, core.ID, definition.VersionView, error) {
	v, gate, err := readWorkflowInstance(ctx, tx, ns, id, false)
	var version definition.VersionView
	if err != nil {
		return v, gate, version, err
	}
	version, err = instanceVersion(ctx, tx, v)
	if err != nil {
		return v, gate, version, err
	}
	if err = authorizeWorkflowQueues(ctx, tx, ns, p, version.Analysis.Queues); err != nil {
		return v, gate, version, err
	}
	v, gate, err = readWorkflowInstance(ctx, tx, ns, id, true)
	return v, gate, version, err
}
func (s *WorkflowInstanceStore) Steps(ctx context.Context, t auth.Token, id core.ID) (instance.StepsView, error) {
	out := instance.StepsView{InstanceID: id, Steps: []instance.StepView{}}
	err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
		v, gate, version, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p)
		if err != nil {
			return err
		}
		out.Revision = v.Revision
		out.Steps, err = readWorkflowStepViews(ctx, tx, t.NamespaceID, id, gate, version, "")
		return err
	})
	return out, err
}
func readWorkflowStepViews(ctx context.Context, tx *sql.Tx, ns, id, gate core.ID, version definition.VersionView, key string) ([]instance.StepView, error) {
	items := []instance.StepView{}
	program, err := engine.Compile(version.Specification)
	if err != nil {
		return nil, err
	}
	rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,COALESCE(a.sequence,0),a.id,a.state,task.id,task.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks task ON task.namespace_id=s.namespace_id AND task.id=a.task_id AND task.owner_kind='workflow_activation' AND task.owner_id=a.id AND task.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.`key`<>? AND (?='' OR s.`key`=?) ORDER BY s.`key` LIMIT 129", gate, ns, id, workflowRootKey, key, key)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var step instance.StepView
		var activation *core.ID
		var task, queue *core.ID
		var state sql.NullString
		if err := rows.Scan(&step.ID, &step.Key, &step.Kind, &step.State, &step.Revision, &step.ActivationID, &step.TaskID, &step.ActivationSequence, &activation, &state, &task, &queue); err != nil {
			return nil, err
		}
		node, ok := program.Step(step.Key)
		if !ok || node.Kind != string(step.Kind) {
			return nil, errors.New("workflow step definition mismatch")
		}
		inactive := step.State == "BLOCKED" || step.State == "SKIPPED"
		if inactive && step.ActivationID != nil || !inactive && (step.ActivationSequence == 0 || step.ActivationID == nil || activation == nil || *activation != *step.ActivationID || state.String != string(step.State)) {
			return nil, errors.New("workflow current activation mismatch")
		}
		if !inactive && (step.Kind == "TASK") != (step.TaskID != nil) {
			return nil, errors.New("workflow direct task ownership mismatch")
		}
		if step.TaskID != nil && (task == nil || queue == nil || *task != *step.TaskID || node.Task == nil || *queue != node.Task.QueueID) {
			return nil, errors.New("workflow task association invalid")
		}
		items = append(items, step)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if key == "" && len(items) != int(version.Analysis.Steps) {
		return nil, errors.New("workflow step set incomplete")
	}
	return items, nil
}