weichenandCursor
f168e44ce7
Activated projects can update duties, approval policy, and document enablement. Changing the workflow definition stays rejected, and project 143 remains frozen. Co-authored-by: Cursor <cursoragent@cursor.com>
1513 lines
47 KiB
Go
1513 lines
47 KiB
Go
package projectinstance
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
||
"chat-agent-system/internal/workflowdef"
|
||
)
|
||
|
||
// 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)
|
||
// GetGoverningInstanceForAccess 供阶段/文档 ABAC：优先 live；否则最新 completed/terminated；
|
||
// 从未激活（无行或仅 planning）返回 (nil, nil)。与 HasLiveInstance（仅 active 系）语义分离。
|
||
GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error)
|
||
HasLiveInstance(ctx context.Context, projectID int64) (bool, error)
|
||
UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error)
|
||
// SavePlanWithRevision 单事务完成 planning upsert + revision bump，
|
||
// 避免 SaveWorkflowPlan 两条独立事务间的版本号与 plan 内容错位。
|
||
SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error)
|
||
ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error)
|
||
ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error)
|
||
CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error)
|
||
TerminateInstance(ctx context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error)
|
||
|
||
SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time, policyRevision *int64) 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, operatorID int64) (*InstanceDetail, error)
|
||
// ReopenStage sets a completed stage back to active without changing other stages.
|
||
ReopenStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) (*InstanceDetail, error)
|
||
MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error
|
||
// UpdateLiveDocTaskDuties 更新 live 实例 config_snapshot 与 document_duty_assignments（单任务）。
|
||
UpdateLiveDocTaskDuties(ctx context.Context, projectID int64, taskKey string, plan WorkflowPlan, assignedBy int64) (*WorkflowInstance, error)
|
||
// UpdateLivePlan 更新 live 快照（职责、审批策略、文档/阶段启用），并同步实例行与职责指派。
|
||
UpdateLivePlan(ctx context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, assignedBy int64) (*WorkflowInstance, 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) GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
|
||
live, err := s.GetLiveInstance(ctx, projectID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if live != nil {
|
||
return live, nil
|
||
}
|
||
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 ('completed', 'terminated')
|
||
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) {
|
||
// Allocate next instance_no so planning drafts after a completed main/rdev
|
||
// instance do not collide on (project_id, instance_no).
|
||
var nextNo int
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT COALESCE(MAX(instance_no), 0) + 1
|
||
FROM project_workflow_instances
|
||
WHERE project_id = $1
|
||
`, inst.ProjectID).Scan(&nextNo); err != nil {
|
||
return nil, err
|
||
}
|
||
inst.InstanceNo = nextNo
|
||
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, $3, 'planning', $4, $5, $6, $7, $8, $9)
|
||
RETURNING id
|
||
`, inst.ProjectID, inst.WorkflowVersionID, nextNo, 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
|
||
}
|
||
|
||
// SavePlanWithRevision 在单事务内完成 planning 实例 upsert 与 plan revision bump。
|
||
// bump 先执行（带期望版本乐观锁），失败则整个事务回滚，保证 plan 内容与版本号原子一致。
|
||
func (s *pgStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, 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, 0, err
|
||
}
|
||
defer tx.Rollback(ctx)
|
||
|
||
// 先 bump revision（乐观锁），冲突即回滚，不写 plan
|
||
var newRev int
|
||
err = tx.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
|
||
`, inst.ProjectID, expectedPlanRevision).Scan(&newRev)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, 0, ErrPlanRevisionConflict
|
||
}
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
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) {
|
||
var nextNo int
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT COALESCE(MAX(instance_no), 0) + 1
|
||
FROM project_workflow_instances
|
||
WHERE project_id = $1
|
||
`, inst.ProjectID).Scan(&nextNo); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
inst.InstanceNo = nextNo
|
||
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, $3, 'planning', $4, $5, $6, $7, $8, $9)
|
||
RETURNING id
|
||
`, inst.ProjectID, inst.WorkflowVersionID, nextNo, inst.ConfigRevision, inst.PlanHash, snapRaw,
|
||
inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
} else if err != nil {
|
||
return nil, 0, 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, 0, err
|
||
}
|
||
}
|
||
|
||
if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return inst, newRev, 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 i, st := range inst.ConfigSnapshot.Stages {
|
||
position := (i + 1) * 10
|
||
var stageID int64
|
||
err = tx.QueryRow(ctx, `
|
||
INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status)
|
||
VALUES ($1, $2, $3, $4, 'pending') RETURNING id
|
||
`, inst.ID, st.StageKey, position, st.Enabled).Scan(&stageID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, dt := range st.DocTasks {
|
||
status := "pending"
|
||
if !dt.Enabled {
|
||
status = "skipped_disabled"
|
||
}
|
||
var docTaskID int64
|
||
err = tx.QueryRow(ctx, `
|
||
INSERT INTO project_doc_task_instances (stage_instance_id, task_key, enabled, status)
|
||
VALUES ($1, $2, $3, $4) RETURNING id
|
||
`, stageID, dt.TaskKey, dt.Enabled, status).Scan(&docTaskID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !dt.Enabled {
|
||
continue
|
||
}
|
||
mode, effective, _ := ResolveDocTaskDuty(dt)
|
||
if mode == DutyModeNone {
|
||
continue
|
||
}
|
||
for role, uid := range map[string]int64{
|
||
DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.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, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
|
||
return s.activateInstanceTx(ctx, s.pool, "", projectID, inst, version, expectedVersion, activatedBy, "")
|
||
}
|
||
|
||
func (s *pgStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
|
||
return s.activateInstanceTx(ctx, s.pool, token, projectID, inst, version, expectedVersion, activatedBy, planHash)
|
||
}
|
||
|
||
func (s *pgStore) activateInstanceTx(ctx context.Context, db pgxDB, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
|
||
workflowKey := ""
|
||
if version != nil {
|
||
workflowKey = version.WorkflowKey
|
||
}
|
||
tx, err := db.Begin(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer tx.Rollback(ctx)
|
||
|
||
usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot)
|
||
var currentPolicyRev int64
|
||
if usesAuthorOnly {
|
||
snap, lockErr, ok := runActivateAuthorOnlyLock(ctx, tx)
|
||
if !ok {
|
||
return nil, ErrAuthorOnlyPolicyUnavailable
|
||
}
|
||
if lockErr != nil {
|
||
return nil, fmt.Errorf("%w: %v", ErrAuthorOnlyPolicyUnavailable, lockErr)
|
||
}
|
||
if !snap.Allowed {
|
||
return nil, ErrAuthorOnlyPolicyDisabled
|
||
}
|
||
currentPolicyRev = snap.Revision
|
||
}
|
||
|
||
if err := runActivateMemberRecheck(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
if token != "" {
|
||
tokenPolicyRev, err := consumeActivationTokenTx(ctx, tx, token, projectID, planHash)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if usesAuthorOnly {
|
||
if tokenPolicyRev == nil || *tokenPolicyRev != currentPolicyRev {
|
||
return nil, ErrAuthorOnlyPolicyChanged
|
||
}
|
||
}
|
||
}
|
||
|
||
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 version != nil {
|
||
if err := s.applyRuntimeMetadata(ctx, tx, inst.ID, version.Definition); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
return activated, nil
|
||
}
|
||
|
||
type pgxDB interface {
|
||
Begin(ctx context.Context) (pgx.Tx, error)
|
||
}
|
||
|
||
func consumeActivationTokenTx(ctx context.Context, tx pgx.Tx, token string, projectID int64, planHash string) (*int64, error) {
|
||
tokenHash := hashActivationToken(token)
|
||
var storedHash string
|
||
var usedAt *time.Time
|
||
var expiresAt time.Time
|
||
var policyRev *int64
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT plan_hash, used_at, expires_at, author_only_policy_revision
|
||
FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2
|
||
`, tokenHash, projectID).Scan(&storedHash, &usedAt, &expiresAt, &policyRev)
|
||
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 := tx.Exec(ctx, `
|
||
UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL
|
||
`, tokenHash)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return nil, ErrTokenInvalid
|
||
}
|
||
return policyRev, nil
|
||
}
|
||
|
||
func (s *pgStore) CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error) {
|
||
if projectID <= 0 {
|
||
return 0, nil
|
||
}
|
||
var count int
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT COUNT(*) FROM task_runs
|
||
WHERE project_id = $1 AND status IN ('pending', 'running')
|
||
`, projectID).Scan(&count)
|
||
if err != nil {
|
||
return 0, nil
|
||
}
|
||
return count, nil
|
||
}
|
||
|
||
func (s *pgStore) applyRuntimeMetadata(ctx context.Context, tx pgx.Tx, instanceID int64, def workflowdef.VersionDefinition) error {
|
||
taskDefs := map[string]workflowdef.DocTaskDef{}
|
||
for _, st := range def.Stages {
|
||
for _, dt := range st.DocTasks {
|
||
taskDefs[dt.Key] = dt
|
||
}
|
||
}
|
||
// 先完整读取并关闭结果集，再执行 UPDATE：
|
||
// pgx 在同一事务连接上存在未读完的 rows 时执行 Exec 会报 "conn busy"。
|
||
rows, err := tx.Query(ctx, `
|
||
SELECT dti.id, dti.task_key
|
||
FROM project_doc_task_instances dti
|
||
JOIN project_stage_instances si ON si.id = dti.stage_instance_id
|
||
WHERE si.workflow_instance_id = $1
|
||
`, instanceID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
type taskInstanceRow struct {
|
||
id int64
|
||
taskKey string
|
||
}
|
||
taskRows := make([]taskInstanceRow, 0)
|
||
for rows.Next() {
|
||
var r taskInstanceRow
|
||
if err := rows.Scan(&r.id, &r.taskKey); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
taskRows = append(taskRows, r)
|
||
}
|
||
iterErr := rows.Err()
|
||
rows.Close()
|
||
if iterErr != nil {
|
||
return iterErr
|
||
}
|
||
for _, r := range taskRows {
|
||
td, ok := taskDefs[r.taskKey]
|
||
if !ok {
|
||
continue
|
||
}
|
||
tplKey := td.TemplateAssetKey
|
||
tplVer := td.TemplateVersion
|
||
if td.Template != nil {
|
||
if td.Template.AssetKey != "" {
|
||
tplKey = td.Template.AssetKey
|
||
}
|
||
if td.Template.Version > 0 {
|
||
tplVer = td.Template.Version
|
||
}
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE project_doc_task_instances SET
|
||
executor_key = $2, quality_gate_key = $3,
|
||
template_asset_key = $4, template_version = $5,
|
||
output_doc_type = $6, output_doc_title = $7, updated_at = now()
|
||
WHERE id = $1
|
||
`, r.id, td.ExecutorKey, td.QualityGateKey, tplKey, tplVer, td.OutputDocType, td.OutputDocTitle); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func hashActivationToken(token string) string {
|
||
sum := sha256.Sum256([]byte(token))
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
func (s *pgStore) SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time, policyRevision *int64) error {
|
||
raw, _ := json.Marshal(snapshot)
|
||
tokenHash := hashActivationToken(token)
|
||
_, err := s.pool.Exec(ctx, `
|
||
INSERT INTO workflow_activation_tokens (token, project_id, plan_hash, expected_project_version, config_snapshot, created_by, expires_at, author_only_policy_revision)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||
`, tokenHash, projectID, planHash, expectedVersion, raw, createdBy, expiresAt, policyRevision)
|
||
return err
|
||
}
|
||
|
||
func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
|
||
tokenHash := hashActivationToken(token)
|
||
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
|
||
`, tokenHash, 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
|
||
`, tokenHash)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
return nil, ErrTokenInvalid
|
||
}
|
||
var plan WorkflowPlan
|
||
_ = json.Unmarshal(snapRaw, &plan)
|
||
return &plan, nil
|
||
}
|
||
|
||
func (s *pgStore) UpdateLiveDocTaskDuties(ctx context.Context, projectID int64, taskKey string, plan WorkflowPlan, assignedBy int64) (*WorkflowInstance, error) {
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer tx.Rollback(ctx)
|
||
|
||
live, err := s.GetLiveInstance(ctx, projectID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if live == nil {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
|
||
planHash, err := computePlanHash(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
snapRaw, err := json.Marshal(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
now := time.Now()
|
||
row := tx.QueryRow(ctx, `
|
||
UPDATE project_workflow_instances SET
|
||
config_snapshot = $2, plan_hash = $3, config_revision = config_revision + 1, updated_at = $4
|
||
WHERE id = $1 AND status IN ('active', 'awaiting_stage_confirmation')
|
||
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
|
||
`, live.ID, snapRaw, planHash, now)
|
||
inst, err := scanInstance(row)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
var docTaskID int64
|
||
err = tx.QueryRow(ctx, `
|
||
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 AND dti.task_key = $2
|
||
`, inst.ID, taskKey).Scan(&docTaskID)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrDocTaskNotFound
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if _, err := tx.Exec(ctx, `DELETE FROM document_duty_assignments WHERE doc_task_instance_id = $1`, docTaskID); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
dt := findDocTaskPlan(plan, taskKey)
|
||
if dt == nil {
|
||
return nil, ErrDocTaskNotFound
|
||
}
|
||
mode, effective, _ := ResolveDocTaskDuty(*dt)
|
||
if mode != DutyModeNone {
|
||
for role, uid := range map[string]int64{
|
||
DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.ApproverUserID,
|
||
} {
|
||
if uid <= 0 {
|
||
continue
|
||
}
|
||
if _, 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, assignedBy); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
return inst, nil
|
||
}
|
||
|
||
func (s *pgStore) UpdateLivePlan(ctx context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, assignedBy int64) (*WorkflowInstance, error) {
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer tx.Rollback(ctx)
|
||
|
||
var instanceID int64
|
||
var revision int
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT id, config_revision
|
||
FROM project_workflow_instances
|
||
WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation')
|
||
FOR UPDATE
|
||
`, projectID).Scan(&instanceID, &revision)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if revision != expectedConfigRevision {
|
||
return nil, ErrInstanceRevisionConflict
|
||
}
|
||
|
||
planHash, err := computePlanHash(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
snapRaw, err := json.Marshal(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
now := time.Now()
|
||
row := tx.QueryRow(ctx, `
|
||
UPDATE project_workflow_instances SET
|
||
config_snapshot = $2, plan_hash = $3, config_revision = config_revision + 1, updated_at = $4
|
||
WHERE id = $1 AND config_revision = $5
|
||
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
|
||
`, instanceID, snapRaw, planHash, now, expectedConfigRevision)
|
||
inst, err := scanInstance(row)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrInstanceRevisionConflict
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for _, st := range plan.Stages {
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE project_stage_instances
|
||
SET enabled = $3, updated_at = $4
|
||
WHERE workflow_instance_id = $1 AND stage_key = $2
|
||
`, inst.ID, st.StageKey, st.Enabled, now); err != nil {
|
||
return nil, err
|
||
}
|
||
for _, dt := range st.DocTasks {
|
||
var docTaskID int64
|
||
err = tx.QueryRow(ctx, `
|
||
UPDATE project_doc_task_instances AS dti SET
|
||
enabled = $3,
|
||
status = CASE
|
||
WHEN $3 = false AND dti.status = 'pending' THEN 'skipped_disabled'
|
||
WHEN $3 = true AND dti.status = 'skipped_disabled' THEN 'pending'
|
||
ELSE dti.status
|
||
END,
|
||
updated_at = $4
|
||
FROM project_stage_instances si
|
||
WHERE dti.stage_instance_id = si.id
|
||
AND si.workflow_instance_id = $1
|
||
AND dti.task_key = $2
|
||
RETURNING dti.id
|
||
`, inst.ID, dt.TaskKey, dt.Enabled, now).Scan(&docTaskID)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrDocTaskNotFound
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := tx.Exec(ctx, `DELETE FROM document_duty_assignments WHERE doc_task_instance_id = $1`, docTaskID); err != nil {
|
||
return nil, err
|
||
}
|
||
if !dt.Enabled || !st.Enabled {
|
||
continue
|
||
}
|
||
mode, effective, _ := ResolveDocTaskDuty(dt)
|
||
if mode == DutyModeNone {
|
||
continue
|
||
}
|
||
for role, uid := range map[string]int64{
|
||
DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.ApproverUserID,
|
||
} {
|
||
if uid <= 0 {
|
||
continue
|
||
}
|
||
if _, 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, assignedBy); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
return inst, nil
|
||
}
|
||
|
||
func (s *pgStore) TerminateInstance(ctx context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error) {
|
||
tx, err := s.pool.Begin(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer tx.Rollback(ctx)
|
||
|
||
now := time.Now()
|
||
row := tx.QueryRow(ctx, `
|
||
UPDATE project_workflow_instances SET status = 'terminated', terminated_at = $3, terminated_by = $4,
|
||
termination_reason = $5, config_revision = config_revision + 1, updated_at = $3
|
||
WHERE id = $1 AND project_id = $2 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
|
||
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
|
||
`, instanceID, projectID, now, terminatedBy, reason)
|
||
inst, err := scanInstance(row)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, ErrTerminationNotAllowed
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
_, err = tx.Exec(ctx, `
|
||
UPDATE projects SET lifecycle_status = 'planning', active_workflow_key = '', updated_at = $2
|
||
WHERE id = $1
|
||
`, projectID, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
_, _ = tx.Exec(ctx, `
|
||
UPDATE task_runs SET status = 'cancelled', updated_at = $2, error_message = 'workflow terminated'
|
||
WHERE project_id = $1 AND status IN ('pending', 'running')
|
||
`, projectID, now)
|
||
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
return inst, 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
|
||
policyRevision *int64
|
||
}
|
||
|
||
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_disabled"
|
||
}
|
||
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 ©Inst, 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 || !IsLiveInstanceStatus(inst.Status) {
|
||
return nil, nil
|
||
}
|
||
copyInst := *inst
|
||
return ©Inst, nil
|
||
}
|
||
|
||
func (m *memStore) GetGoverningInstanceForAccess(_ context.Context, projectID int64) (*WorkflowInstance, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
inst := m.live[projectID]
|
||
if inst == nil {
|
||
return nil, nil
|
||
}
|
||
if IsLiveInstanceStatus(inst.Status) || IsClosedInstanceStatus(inst.Status) {
|
||
copyInst := *inst
|
||
return ©Inst, nil
|
||
}
|
||
// planning 等非治理态：视为从未激活
|
||
return nil, nil
|
||
}
|
||
|
||
func (m *memStore) HasLiveInstance(_ context.Context, projectID int64) (bool, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
inst := m.live[projectID]
|
||
return inst != nil && IsLiveInstanceStatus(inst.Status), 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] = ©Inst
|
||
m.rebuildStagesLocked(©Inst, false)
|
||
return ©Inst, nil
|
||
}
|
||
|
||
// SavePlanWithRevision 内存实现：bump 与 upsert 在同一互斥锁内，天然原子。
|
||
func (m *memStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) {
|
||
newRev, err := m.BumpPlanRevision(ctx, inst.ProjectID, expectedPlanRevision)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
saved, err := m.UpsertPlanningInstance(ctx, inst)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
return saved, newRev, nil
|
||
}
|
||
|
||
func (m *memStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
|
||
usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot)
|
||
var currentPolicyRev int64
|
||
if usesAuthorOnly {
|
||
snap, lockErr, ok := runActivateAuthorOnlyLock(ctx, nil)
|
||
if !ok {
|
||
return nil, ErrAuthorOnlyPolicyUnavailable
|
||
}
|
||
if lockErr != nil {
|
||
return nil, fmt.Errorf("%w: %v", ErrAuthorOnlyPolicyUnavailable, lockErr)
|
||
}
|
||
if !snap.Allowed {
|
||
return nil, ErrAuthorOnlyPolicyDisabled
|
||
}
|
||
currentPolicyRev = snap.Revision
|
||
}
|
||
if err := runActivateMemberRecheck(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
tokenPolicyRev, err := m.consumeActivationTokenLocked(ctx, token, projectID, planHash)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if usesAuthorOnly {
|
||
if tokenPolicyRev == nil || *tokenPolicyRev != currentPolicyRev {
|
||
return nil, ErrAuthorOnlyPolicyChanged
|
||
}
|
||
}
|
||
return m.ActivateInstance(ctx, projectID, inst, version, expectedVersion, activatedBy)
|
||
}
|
||
|
||
func (m *memStore) CountActiveTaskRuns(_ context.Context, _ int64) (int, error) {
|
||
return 0, nil
|
||
}
|
||
|
||
func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
|
||
workflowKey := ""
|
||
if version != nil {
|
||
workflowKey = version.WorkflowKey
|
||
}
|
||
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, _ int64) (*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
|
||
anotherActive := false
|
||
for i, rec := range records {
|
||
if i != activeIdx && rec.enabled && rec.status == StageActive {
|
||
anotherActive = true
|
||
break
|
||
}
|
||
}
|
||
nextIdx := -1
|
||
if !anotherActive {
|
||
for i, rec := range records {
|
||
if rec.enabled && rec.status == StagePending {
|
||
nextIdx = i
|
||
break
|
||
}
|
||
}
|
||
}
|
||
inst.ConfigRevision++
|
||
inst.UpdatedAt = time.Now()
|
||
if anotherActive {
|
||
inst.Status = InstanceActive
|
||
} else if nextIdx < 0 {
|
||
inst.Status = InstanceCompleted
|
||
if pm := m.meta[inst.ProjectID]; pm.lifecycle != "" {
|
||
pm.lifecycle = LifecyclePlanning
|
||
pm.activeKey = ""
|
||
m.meta[inst.ProjectID] = pm
|
||
}
|
||
} else {
|
||
records[nextIdx].status = StageActive
|
||
}
|
||
m.stages[instanceID] = records
|
||
copyInst := *inst
|
||
m.live[inst.ProjectID] = ©Inst
|
||
return m.instanceDetailLocked(©Inst), nil
|
||
}
|
||
|
||
func (m *memStore) ReopenStage(_ context.Context, instanceID int64, stageKey string, docTaskKeys []string, _ int64) (*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
|
||
}
|
||
records := m.stages[instanceID]
|
||
idx := -1
|
||
for i, rec := range records {
|
||
if rec.stageKey == stageKey {
|
||
idx = i
|
||
break
|
||
}
|
||
}
|
||
if idx < 0 {
|
||
return nil, ErrInvalidStage
|
||
}
|
||
if records[idx].status != StageCompleted {
|
||
return nil, ErrStageNotCompleted
|
||
}
|
||
records[idx].status = StageActive
|
||
wanted := map[string]bool{}
|
||
for _, key := range docTaskKeys {
|
||
wanted[key] = true
|
||
}
|
||
for j := range records[idx].tasks {
|
||
if wanted[records[idx].tasks[j].taskKey] {
|
||
records[idx].tasks[j].status = "in_progress"
|
||
}
|
||
}
|
||
if inst.Status == InstanceCompleted {
|
||
inst.Status = InstanceActive
|
||
}
|
||
inst.ConfigRevision++
|
||
inst.UpdatedAt = time.Now()
|
||
m.stages[instanceID] = records
|
||
copyInst := *inst
|
||
m.live[inst.ProjectID] = ©Inst
|
||
return m.instanceDetailLocked(©Inst), 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) UpdateLiveDocTaskDuties(_ context.Context, projectID int64, taskKey string, plan WorkflowPlan, _ int64) (*WorkflowInstance, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
live := m.live[projectID]
|
||
if live == nil {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
if findDocTaskPlan(plan, taskKey) == nil {
|
||
return nil, ErrDocTaskNotFound
|
||
}
|
||
planHash, err := computePlanHash(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
copyInst := *live
|
||
copyInst.ConfigSnapshot = plan
|
||
copyInst.PlanHash = planHash
|
||
copyInst.ConfigRevision++
|
||
copyInst.UpdatedAt = time.Now()
|
||
m.live[projectID] = ©Inst
|
||
return ©Inst, nil
|
||
}
|
||
|
||
func (m *memStore) UpdateLivePlan(_ context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, _ int64) (*WorkflowInstance, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
live := m.live[projectID]
|
||
if live == nil {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
if live.ConfigRevision != expectedConfigRevision {
|
||
return nil, ErrInstanceRevisionConflict
|
||
}
|
||
planHash, err := computePlanHash(plan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
copyInst := *live
|
||
copyInst.ConfigSnapshot = plan
|
||
copyInst.PlanHash = planHash
|
||
copyInst.ConfigRevision++
|
||
copyInst.UpdatedAt = time.Now()
|
||
m.live[projectID] = ©Inst
|
||
|
||
records := m.stages[copyInst.ID]
|
||
for i := range records {
|
||
for _, st := range plan.Stages {
|
||
if records[i].stageKey == st.StageKey {
|
||
records[i].enabled = st.Enabled
|
||
}
|
||
}
|
||
for j := range records[i].tasks {
|
||
dt := findDocTaskPlan(plan, records[i].tasks[j].taskKey)
|
||
if dt == nil {
|
||
continue
|
||
}
|
||
records[i].tasks[j].enabled = dt.Enabled
|
||
if !dt.Enabled && records[i].tasks[j].status == "pending" {
|
||
records[i].tasks[j].status = "skipped_disabled"
|
||
}
|
||
if dt.Enabled && records[i].tasks[j].status == "skipped_disabled" {
|
||
records[i].tasks[j].status = "pending"
|
||
}
|
||
}
|
||
}
|
||
m.stages[copyInst.ID] = records
|
||
return ©Inst, nil
|
||
}
|
||
|
||
func (m *memStore) TerminateInstance(_ context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
inst := m.live[projectID]
|
||
if inst == nil || inst.ID != instanceID {
|
||
return nil, ErrInstanceNotFound
|
||
}
|
||
now := time.Now()
|
||
inst.Status = InstanceTerminated
|
||
copyInst := *inst
|
||
m.live[projectID] = ©Inst
|
||
pm := m.ensureMeta(projectID)
|
||
pm.lifecycle = LifecyclePlanning
|
||
pm.activeKey = ""
|
||
m.meta[projectID] = pm
|
||
_ = terminatedBy
|
||
_ = reason
|
||
_ = now
|
||
return ©Inst, nil
|
||
}
|
||
|
||
func (m *memStore) SaveActivationToken(_ context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, _ int64, expiresAt time.Time, policyRevision *int64) error {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
var revCopy *int64
|
||
if policyRevision != nil {
|
||
v := *policyRevision
|
||
revCopy = &v
|
||
}
|
||
m.tokens[hashActivationToken(token)] = activationTokenRec{
|
||
projectID: projectID, planHash: planHash, expectedVer: expectedVersion,
|
||
snapshot: snapshot, expiresAt: expiresAt, policyRevision: revCopy,
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (m *memStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
|
||
_, plan, err := m.consumeActivationTokenWithRev(ctx, token, projectID, planHash)
|
||
return plan, err
|
||
}
|
||
|
||
func (m *memStore) consumeActivationTokenLocked(ctx context.Context, token string, projectID int64, planHash string) (*int64, error) {
|
||
rev, _, err := m.consumeActivationTokenWithRev(ctx, token, projectID, planHash)
|
||
return rev, err
|
||
}
|
||
|
||
func (m *memStore) consumeActivationTokenWithRev(_ context.Context, token string, projectID int64, planHash string) (*int64, *WorkflowPlan, error) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
tokenHash := hashActivationToken(token)
|
||
rec, ok := m.tokens[tokenHash]
|
||
if !ok || rec.used || rec.projectID != projectID || rec.planHash != planHash || time.Now().After(rec.expiresAt) {
|
||
return nil, nil, ErrTokenInvalid
|
||
}
|
||
rec.used = true
|
||
m.tokens[tokenHash] = rec
|
||
plan := rec.snapshot
|
||
return rec.policyRevision, &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
|
||
}
|