orbit/internal/storage/mysql/workflow_runtime_graph.go

package mysql

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

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

// lockWorkflowGraph 只由内部运行器调用。身份→应用额度→排序队列→实例→门控，
// 不用启动凭据继续授权；停用身份仍收集在途结果，但禁止创建新工作。
func lockWorkflowGraph(ctx context.Context, tx *sql.Tx, ns, id core.ID, skip bool) (*workflowGraph, bool, error) {
	var enabled bool
	if err := tx.QueryRowContext(ctx, `SELECT n.enabled AND a.enabled FROM orbit_namespaces n JOIN orbit_applications a ON a.id=n.application_id WHERE n.id=? FOR SHARE`, ns).Scan(&enabled); err != nil {
		return nil, false, err
	}
	if ok, err := lockApplicationResources(ctx, tx, ns, skip); err != nil || !ok {
		return nil, false, err
	}
	v, gate, err := readWorkflowInstance(ctx, tx, ns, id, false)
	if err != nil {
		return nil, false, err
	}
	version, err := instanceVersion(ctx, tx, v)
	if err != nil {
		return nil, false, err
	}
	for _, queue := range version.Analysis.Queues {
		if _, _, err := lockQueueAfterApplication(ctx, tx, ns, queue); err != nil {
			return nil, false, err
		}
	}
	var locked core.ID
	if err := tx.QueryRowContext(ctx, `SELECT id FROM weave_instances WHERE namespace_id=? AND id=? FOR UPDATE`, ns, id).Scan(&locked); err != nil {
		return nil, false, err
	}
	v, currentGate, err := readWorkflowInstance(ctx, tx, ns, id, false)
	if err != nil {
		return nil, false, err
	}
	if currentGate != gate || v.DefinitionVersionID != version.ID {
		return nil, false, errors.New("workflow identity changed")
	}
	var gatePaused bool
	if err := tx.QueryRowContext(ctx, `SELECT paused FROM pulse_execution_gates WHERE namespace_id=? AND id=? FOR UPDATE`, ns, gate).Scan(&gatePaused); err != nil {
		return nil, false, err
	}
	program, err := engine.Compile(version.Specification)
	if err != nil {
		return nil, false, err
	}
	g := &workflowGraph{view: v, gate: gate, version: version, program: program, steps: map[string]*workflowStep{}}
	rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,a.id,a.state,a.input,a.output,t.id,t.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 t ON t.namespace_id=s.namespace_id AND t.id=a.task_id AND t.owner_kind='workflow_activation' AND t.owner_id=a.id AND t.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? ORDER BY s.`key` LIMIT 130", gate, ns, id)
	if err != nil {
		return nil, false, err
	}
	defer rows.Close()
	for rows.Next() {
		row := &workflowStep{}
		var activation, task, queue *core.ID
		var state sql.NullString
		var input, output []byte
		if err := rows.Scan(&row.view.ID, &row.view.Key, &row.view.Kind, &row.view.State, &row.view.Revision, &row.view.ActivationID, &row.view.TaskID, &activation, &state, &input, &output, &task, &queue); err != nil {
			return nil, false, err
		}
		row.input, row.output = input, output
		key := row.view.Key
		node, ok := program.Step(key)
		if key == workflowRootKey {
			if row.view.Kind != weave.NodeJoin {
				return nil, false, errors.New("workflow root kind invalid")
			}
		} else if !ok || node.Kind != string(row.view.Kind) {
			return nil, false, errors.New("workflow step definition mismatch")
		}
		inactive := row.view.State == weave.StepBlocked || row.view.State == weave.StepSkipped
		if inactive && row.view.ActivationID != nil || !inactive && (row.view.ActivationID == nil || activation == nil || *activation != *row.view.ActivationID || state.String != string(row.view.State)) {
			return nil, false, errors.New("workflow current activation mismatch")
		}
		if !inactive && (row.view.Kind == weave.NodeTask) != (row.view.TaskID != nil) {
			return nil, false, errors.New("workflow task ownership mismatch")
		}
		if row.view.TaskID != nil && (task == nil || queue == nil || *task != *row.view.TaskID || node.Task == nil || *queue != node.Task.QueueID) {
			return nil, false, errors.New("workflow task association invalid")
		}
		g.steps[key] = row
	}
	if err := rows.Err(); err != nil {
		return nil, false, err
	}
	root := g.steps[workflowRootKey]
	if len(g.steps) != int(version.Analysis.Steps)+1 || root == nil || root.view.ActivationID == nil {
		return nil, false, errors.New("workflow step set incomplete")
	}
	allow := enabled && !gatePaused && !v.Paused && v.CancelRequestedAt == nil && (v.State == weave.InstanceRunning || v.State == weave.InstanceWaiting)
	return g, allow, nil
}

func workflowSnapshot(g *workflowGraph, allow bool) engine.Snapshot {
	s := engine.Snapshot{WorkflowID: g.view.ID, BusinessKey: g.view.BusinessKey, Input: g.view.Input, Steps: map[string]engine.State{}, AllowNew: allow}
	for key, row := range g.steps {
		if key != workflowRootKey {
			s.Steps[key] = engine.State{Status: row.view.State, Input: row.input, Output: row.output}
		}
	}
	return s
}

// saveWorkflowAdvance 保留根步骤的最新版本；重新调度不重置业务截止或任务预算。
func saveWorkflowAdvance(ctx context.Context, tx *sql.Tx, g *workflowGraph, due time.Time) error {
	id, err := core.NewID()
	if err != nil {
		return err
	}
	root := g.steps[workflowRootKey]
	_, err = tx.ExecContext(ctx, `INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at) VALUES (?,?,?,?,?,'{"schema_version":1,"kind":"ROOT"}',?) ON DUPLICATE KEY UPDATE expected_step_revision=VALUES(expected_step_revision),checkpoint=VALUES(checkpoint),available_at=VALUES(available_at),revision=revision+1,updated_at=UTC_TIMESTAMP(6)`, g.view.NamespaceID, id, g.view.ID, root.view.ID, root.view.Revision, due)
	return err
}