Agent-System/internal/projectinstance/store.go

package projectinstance

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

// TablesReady 检查 Phase 3 表是否已迁移。
func TablesReady(ctx context.Context, pool *pgxpool.Pool) bool {
	if pool == nil {
		return false
	}
	var ok bool
	err := pool.QueryRow(ctx, `
		SELECT EXISTS (
			SELECT 1 FROM information_schema.tables
			WHERE table_schema = 'public' AND table_name = 'project_workflow_instances'
		)
	`).Scan(&ok)
	return err == nil && ok
}

// Store 项目实例持久化接口。
type Store interface {
	GetProjectMeta(ctx context.Context, projectID int64) (lifecycleStatus string, activeWorkflowKey string, selectionVersion int, planRevision int, err error)
	UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error
	BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error)

	GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
	GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
	HasLiveInstance(ctx context.Context, projectID int64) (bool, error)
	UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error)
	ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error)

	SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error
	ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error)

	GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error)
	AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error)
	MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error
}

// NewStore 构建 PG 或内存 Store。
func NewStore(pool *pgxpool.Pool) Store {
	if pool != nil && TablesReady(context.Background(), pool) {
		return &pgStore{pool: pool}
	}
	return newMemStore()
}

type pgStore struct {
	pool *pgxpool.Pool
}

func scanInstance(row pgx.Row) (*WorkflowInstance, error) {
	var inst WorkflowInstance
	var snapRaw []byte
	var activatedBy, createdBy *int64
	err := row.Scan(
		&inst.ID, &inst.ProjectID, &inst.WorkflowVersionID, &inst.InstanceNo, &inst.Status,
		&inst.ConfigRevision, &inst.PlanHash, &snapRaw, &activatedBy, &inst.ActivatedAt,
		&createdBy, &inst.CreatedAt, &inst.UpdatedAt,
	)
	if err != nil {
		return nil, err
	}
	inst.ActivatedBy = activatedBy
	inst.CreatedBy = createdBy
	if len(snapRaw) > 0 {
		_ = json.Unmarshal(snapRaw, &inst.ConfigSnapshot)
	}
	return &inst, nil
}

func (s *pgStore) GetProjectMeta(ctx context.Context, projectID int64) (string, string, int, int, error) {
	var lifecycle, activeKey string
	var selVer, planRev int
	err := s.pool.QueryRow(ctx, `
		SELECT COALESCE(lifecycle_status, 'planning'), COALESCE(active_workflow_key, ''),
		       COALESCE(workflow_selection_version, 1), COALESCE(workflow_plan_revision, 0)
		FROM projects WHERE id = $1
	`, projectID).Scan(&lifecycle, &activeKey, &selVer, &planRev)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", "", 0, 0, ErrProjectNotFound
	}
	return lifecycle, activeKey, selVer, planRev, err
}

func (s *pgStore) UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error {
	tag, err := s.pool.Exec(ctx, `
		UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
		       workflow_selection_version = workflow_selection_version + 1,
		       active_workflow_changed_at = now(), active_workflow_changed_by = $4,
		       updated_at = now()
		WHERE id = $1 AND workflow_selection_version = $3
	`, projectID, workflowKey, expectedVersion, activatedBy)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrConcurrentConflict
	}
	return nil
}

func (s *pgStore) BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error) {
	var newRev int
	err := s.pool.QueryRow(ctx, `
		UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now()
		WHERE id = $1 AND workflow_plan_revision = $2
		RETURNING workflow_plan_revision
	`, projectID, expectedPlanRevision).Scan(&newRev)
	if errors.Is(err, pgx.ErrNoRows) {
		return 0, ErrPlanRevisionConflict
	}
	return newRev, err
}

func (s *pgStore) GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
	row := s.pool.QueryRow(ctx, `
		SELECT id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
		       config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
		FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning'
		ORDER BY id DESC LIMIT 1
	`, projectID)
	inst, err := scanInstance(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	return inst, err
}

func (s *pgStore) GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
	row := s.pool.QueryRow(ctx, `
		SELECT id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
		       config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
		FROM project_workflow_instances
		WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
		ORDER BY id DESC LIMIT 1
	`, projectID)
	inst, err := scanInstance(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	return inst, err
}

func (s *pgStore) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) {
	var ok bool
	err := s.pool.QueryRow(ctx, `
		SELECT EXISTS (
			SELECT 1 FROM project_workflow_instances
			WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
		)
	`, projectID).Scan(&ok)
	return ok, err
}

func (s *pgStore) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
	snapRaw, _ := json.Marshal(inst.ConfigSnapshot)
	now := time.Now()
	if inst.CreatedAt.IsZero() {
		inst.CreatedAt = now
	}
	inst.UpdatedAt = now

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	var existingID int64
	err = tx.QueryRow(ctx, `
		SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1
	`, inst.ProjectID).Scan(&existingID)
	if errors.Is(err, pgx.ErrNoRows) {
		err = tx.QueryRow(ctx, `
			INSERT INTO project_workflow_instances (
				project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
				config_snapshot, created_by, created_at, updated_at
			) VALUES ($1, $2, 1, 'planning', $3, $4, $5, $6, $7, $8)
			RETURNING id
		`, inst.ProjectID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw,
			inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
		if err != nil {
			return nil, err
		}
	} else if err != nil {
		return nil, err
	} else {
		err = tx.QueryRow(ctx, `
			UPDATE project_workflow_instances SET
				workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5,
				updated_at = $6
			WHERE id = $1 AND status = 'planning'
			RETURNING id
		`, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID)
		if err != nil {
			return nil, err
		}
	}

	if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
		return nil, err
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return inst, nil
}

func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst *WorkflowInstance) error {
	_, err := tx.Exec(ctx, `
		DELETE FROM document_duty_assignments WHERE doc_task_instance_id IN (
			SELECT dti.id FROM project_doc_task_instances dti
			JOIN project_stage_instances si ON si.id = dti.stage_instance_id
			WHERE si.workflow_instance_id = $1
		)
	`, inst.ID)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `
		DELETE FROM project_doc_task_instances WHERE stage_instance_id IN (
			SELECT id FROM project_stage_instances WHERE workflow_instance_id = $1
		)
	`, inst.ID)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `DELETE FROM project_stage_instances WHERE workflow_instance_id = $1`, inst.ID)
	if err != nil {
		return err
	}

	for _, st := range inst.ConfigSnapshot.Stages {
		var stageID int64
		err = tx.QueryRow(ctx, `
			INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status)
			VALUES ($1, $2, 0, $3, 'pending') RETURNING id
		`, inst.ID, st.StageKey, st.Enabled).Scan(&stageID)
		if err != nil {
			return err
		}
		for _, dt := range st.DocTasks {
			var docTaskID int64
			err = tx.QueryRow(ctx, `
				INSERT INTO project_doc_task_instances (stage_instance_id, task_key, enabled, status)
				VALUES ($1, $2, $3, 'pending') RETURNING id
			`, stageID, dt.TaskKey, dt.Enabled).Scan(&docTaskID)
			if err != nil {
				return err
			}
			if !dt.Enabled {
				continue
			}
			for role, uid := range map[string]int64{
				DutyAuthor: dt.Duties.AuthorUserID, DutyReviewer: dt.Duties.ReviewerUserID, DutyApprover: dt.Duties.ApproverUserID,
			} {
				if uid <= 0 {
					continue
				}
				_, err = tx.Exec(ctx, `
					INSERT INTO document_duty_assignments (doc_task_instance_id, duty_role, user_id, assigned_by)
					VALUES ($1, $2, $3, $4)
				`, docTaskID, role, uid, inst.CreatedBy)
				if err != nil {
					return err
				}
			}
		}
	}
	return nil
}

func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	tag, err := tx.Exec(ctx, `
		UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
		       workflow_selection_version = workflow_selection_version + 1,
		       active_workflow_changed_at = now(), active_workflow_changed_by = $4,
		       updated_at = now()
		WHERE id = $1 AND workflow_selection_version = $3
	`, projectID, workflowKey, expectedVersion, activatedBy)
	if err != nil {
		return nil, err
	}
	if tag.RowsAffected() == 0 {
		return nil, ErrConcurrentConflict
	}

	now := time.Now()
	row := tx.QueryRow(ctx, `
		UPDATE project_workflow_instances SET status = 'active', activated_by = $2, activated_at = $3, updated_at = $3
		WHERE id = $1 AND status = 'planning'
		RETURNING id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
		          config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
	`, inst.ID, activatedBy, now)
	activated, err := scanInstance(row)
	if err != nil {
		return nil, err
	}

	_, err = tx.Exec(ctx, `
		UPDATE project_stage_instances SET status = CASE WHEN enabled THEN 'active' ELSE 'skipped' END, updated_at = now()
		WHERE workflow_instance_id = $1 AND stage_key = (
			SELECT stage_key FROM project_stage_instances
			WHERE workflow_instance_id = $1 AND enabled = true
			ORDER BY id ASC LIMIT 1
		)
	`, inst.ID)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `
		UPDATE project_stage_instances SET status = 'pending', updated_at = now()
		WHERE workflow_instance_id = $1 AND status <> 'active' AND enabled = true
	`, inst.ID)
	if err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return activated, nil
}

func (s *pgStore) SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error {
	raw, _ := json.Marshal(snapshot)
	_, err := s.pool.Exec(ctx, `
		INSERT INTO workflow_activation_tokens (token, project_id, plan_hash, expected_project_version, config_snapshot, created_by, expires_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7)
	`, token, projectID, planHash, expectedVersion, raw, createdBy, expiresAt)
	return err
}

func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
	var snapRaw []byte
	var storedHash string
	var usedAt *time.Time
	var expiresAt time.Time
	err := s.pool.QueryRow(ctx, `
		SELECT plan_hash, config_snapshot, used_at, expires_at
		FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2
	`, token, projectID).Scan(&storedHash, &snapRaw, &usedAt, &expiresAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrTokenInvalid
	}
	if err != nil {
		return nil, err
	}
	if usedAt != nil || time.Now().After(expiresAt) || storedHash != planHash {
		return nil, ErrTokenInvalid
	}
	tag, err := s.pool.Exec(ctx, `
		UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL
	`, token)
	if err != nil {
		return nil, err
	}
	if tag.RowsAffected() == 0 {
		return nil, ErrTokenInvalid
	}
	var plan WorkflowPlan
	_ = json.Unmarshal(snapRaw, &plan)
	return &plan, nil
}

// =============================================================================
// 内存 Store（单元测试 / 无 PG 迁移）
// =============================================================================

type memStageRecord struct {
	id       int64
	stageKey string
	position int
	enabled  bool
	status   string
	tasks    []memDocTaskRecord
}

type memDocTaskRecord struct {
	id      int64
	taskKey string
	enabled bool
	status  string
}

type memStore struct {
	mu           sync.Mutex
	meta         map[int64]projectMeta
	plan         map[int64]*WorkflowInstance
	live         map[int64]*WorkflowInstance
	tokens       map[string]activationTokenRec
	stages       map[int64][]memStageRecord
	nextStageID  int64
	nextTaskID   int64
}

type projectMeta struct {
	lifecycle     string
	activeKey     string
	selectionVer  int
	planRevision  int
}

type activationTokenRec struct {
	projectID      int64
	planHash       string
	expectedVer    int
	snapshot       WorkflowPlan
	expiresAt      time.Time
	used           bool
}

func newMemStore() *memStore {
	return &memStore{
		meta:        make(map[int64]projectMeta),
		plan:        make(map[int64]*WorkflowInstance),
		live:        make(map[int64]*WorkflowInstance),
		tokens:      make(map[string]activationTokenRec),
		stages:      make(map[int64][]memStageRecord),
		nextStageID: 1,
		nextTaskID:  1,
	}
}

func (m *memStore) rebuildStagesLocked(inst *WorkflowInstance, activate bool) {
	records := make([]memStageRecord, 0, len(inst.ConfigSnapshot.Stages))
	firstActive := true
	for i, st := range inst.ConfigSnapshot.Stages {
		rec := memStageRecord{
			id: m.nextStageID, stageKey: st.StageKey, position: i, enabled: st.Enabled,
		}
		m.nextStageID++
		if !st.Enabled {
			rec.status = StageSkipped
		} else if activate && firstActive {
			rec.status = StageActive
			firstActive = false
		} else {
			rec.status = StagePending
		}
		for _, dt := range st.DocTasks {
			task := memDocTaskRecord{id: m.nextTaskID, taskKey: dt.TaskKey, enabled: dt.Enabled, status: "pending"}
			m.nextTaskID++
			if !dt.Enabled {
				task.status = "skipped"
			}
			rec.tasks = append(rec.tasks, task)
		}
		records = append(records, rec)
	}
	m.stages[inst.ID] = records
}

func (m *memStore) instanceDetailLocked(inst *WorkflowInstance) *InstanceDetail {
	detail := &InstanceDetail{Instance: *inst, Stages: make([]StageInstance, 0)}
	for _, rec := range m.stages[inst.ID] {
		st := StageInstance{
			ID: rec.id, StageKey: rec.stageKey, Position: rec.position, Enabled: rec.enabled, Status: rec.status,
		}
		if st.Status == StageActive {
			detail.ActiveStageKey = st.StageKey
		}
		for _, t := range rec.tasks {
			st.DocTasks = append(st.DocTasks, DocTaskInstance{
				ID: t.id, TaskKey: t.taskKey, Enabled: t.enabled, Status: t.status, Approved: t.status == "approved",
			})
		}
		detail.Stages = append(detail.Stages, st)
	}
	return detail
}

func (m *memStore) ensureMeta(projectID int64) projectMeta {
	if pm, ok := m.meta[projectID]; ok {
		return pm
	}
	pm := projectMeta{lifecycle: LifecyclePlanning, selectionVer: 1}
	m.meta[projectID] = pm
	return pm
}

func (m *memStore) GetProjectMeta(_ context.Context, projectID int64) (string, string, int, int, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	pm, ok := m.meta[projectID]
	if !ok {
		return "", "", 0, 0, ErrProjectNotFound
	}
	return pm.lifecycle, pm.activeKey, pm.selectionVer, pm.planRevision, nil
}

func (m *memStore) UpdateProjectAfterActivate(_ context.Context, projectID int64, workflowKey string, expectedVersion int, _ int64) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	pm := m.ensureMeta(projectID)
	if pm.selectionVer != expectedVersion {
		return ErrConcurrentConflict
	}
	pm.lifecycle = LifecycleActive
	pm.activeKey = workflowKey
	pm.selectionVer++
	m.meta[projectID] = pm
	return nil
}

func (m *memStore) BumpPlanRevision(_ context.Context, projectID int64, expectedPlanRevision int) (int, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	pm := m.ensureMeta(projectID)
	if pm.planRevision != expectedPlanRevision {
		return 0, ErrPlanRevisionConflict
	}
	pm.planRevision++
	m.meta[projectID] = pm
	return pm.planRevision, nil
}

func (m *memStore) GetPlanningInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	inst := m.plan[projectID]
	if inst == nil {
		return nil, nil
	}
	copyInst := *inst
	return &copyInst, nil
}

func (m *memStore) GetLiveInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	inst := m.live[projectID]
	if inst == nil {
		return nil, nil
	}
	copyInst := *inst
	return &copyInst, nil
}

func (m *memStore) HasLiveInstance(_ context.Context, projectID int64) (bool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.live[projectID] != nil, nil
}

func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if inst.ID == 0 {
		inst.ID = int64(len(m.plan) + 1)
	}
	inst.Status = InstancePlanning
	now := time.Now()
	if inst.CreatedAt.IsZero() {
		inst.CreatedAt = now
	}
	inst.UpdatedAt = now
	copyInst := *inst
	m.plan[inst.ProjectID] = &copyInst
	m.rebuildStagesLocked(&copyInst, false)
	return &copyInst, nil
}

func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	pm := m.ensureMeta(projectID)
	if pm.selectionVer != expectedVersion {
		return nil, ErrConcurrentConflict
	}
	pm.lifecycle = LifecycleActive
	pm.activeKey = workflowKey
	pm.selectionVer++
	m.meta[projectID] = pm

	now := time.Now()
	activated := *inst
	activated.Status = InstanceActive
	activated.ActivatedBy = &activatedBy
	activated.ActivatedAt = &now
	m.live[projectID] = &activated
	m.rebuildStagesLocked(&activated, true)
	delete(m.plan, projectID)
	return &activated, nil
}

func (m *memStore) GetInstanceByID(_ context.Context, instanceID int64) (*InstanceDetail, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, inst := range m.live {
		if inst.ID == instanceID {
			return m.instanceDetailLocked(inst), nil
		}
	}
	for _, inst := range m.plan {
		if inst.ID == instanceID {
			return m.instanceDetailLocked(inst), nil
		}
	}
	return nil, ErrInstanceNotFound
}

func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int) (*InstanceDetail, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	var inst *WorkflowInstance
	for _, live := range m.live {
		if live.ID == instanceID {
			inst = live
			break
		}
	}
	if inst == nil {
		return nil, ErrInstanceNotFound
	}
	if inst.ConfigRevision != expectedRevision {
		return nil, ErrInstanceRevisionConflict
	}
	records := m.stages[instanceID]
	activeIdx := -1
	for i, rec := range records {
		if rec.stageKey == stageKey {
			if rec.status != StageActive {
				return nil, ErrStageNotActive
			}
			activeIdx = i
			break
		}
	}
	if activeIdx < 0 {
		return nil, ErrInvalidStage
	}
	records[activeIdx].status = StageCompleted
	for j := range records[activeIdx].tasks {
		if records[activeIdx].tasks[j].enabled {
			records[activeIdx].tasks[j].status = "approved"
		}
	}
	nextIdx := -1
	for i, rec := range records {
		if rec.enabled && rec.status == StagePending {
			nextIdx = i
			break
		}
	}
	inst.ConfigRevision++
	inst.UpdatedAt = time.Now()
	if nextIdx < 0 {
		inst.Status = InstanceCompleted
	} else {
		records[nextIdx].status = StageActive
	}
	m.stages[instanceID] = records
	copyInst := *inst
	m.live[inst.ProjectID] = &copyInst
	return m.instanceDetailLocked(&copyInst), nil
}

func (m *memStore) MarkDocTaskApproved(_ context.Context, projectID int64, taskKey string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	live := m.live[projectID]
	if live == nil {
		return nil
	}
	records := m.stages[live.ID]
	for i := range records {
		for j := range records[i].tasks {
			if records[i].tasks[j].taskKey == taskKey && records[i].tasks[j].enabled {
				records[i].tasks[j].status = "approved"
			}
		}
	}
	m.stages[live.ID] = records
	return nil
}

func (m *memStore) SaveActivationToken(_ context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, _ int64, expiresAt time.Time) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.tokens[token] = activationTokenRec{
		projectID: projectID, planHash: planHash, expectedVer: expectedVersion,
		snapshot: snapshot, expiresAt: expiresAt,
	}
	return nil
}

func (m *memStore) ConsumeActivationToken(_ context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	rec, ok := m.tokens[token]
	if !ok || rec.used || rec.projectID != projectID || rec.planHash != planHash || time.Now().After(rec.expiresAt) {
		return nil, ErrTokenInvalid
	}
	rec.used = true
	m.tokens[token] = rec
	plan := rec.snapshot
	return &plan, nil
}

// SeedProjectMeta 供测试注入项目元数据。
func SeedProjectMeta(store Store, projectID int64, lifecycle, activeKey string, selectionVer, planRev int) error {
	ms, ok := store.(*memStore)
	if !ok {
		return fmt.Errorf("SeedProjectMeta only supported on memStore")
	}
	ms.mu.Lock()
	defer ms.mu.Unlock()
	ms.meta[projectID] = projectMeta{
		lifecycle: lifecycle, activeKey: activeKey,
		selectionVer: selectionVer, planRevision: planRev,
	}
	return nil
}