diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index c6aa20e3..f5e2fcf9 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -2,18 +2,12 @@ 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/taskrun" "chat-agent-system/internal/workflowdef" ) @@ -98,1415 +92,3 @@ func scanInstance(row pgx.Row) (*WorkflowInstance, error) { } 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 - } - - if err := taskrun.CancelOpenByProject(ctx, tx, projectID, now, "workflow terminated"); err != nil { - return nil, err - } - - 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 -} diff --git a/internal/projectinstance/store_mem.go b/internal/projectinstance/store_mem.go new file mode 100644 index 00000000..d8e420a7 --- /dev/null +++ b/internal/projectinstance/store_mem.go @@ -0,0 +1,579 @@ +package projectinstance + +import ( + "context" + "fmt" + "sync" + "time" + + "chat-agent-system/internal/workflowdef" +) + +// ============================================================================= +// 内存 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 +} diff --git a/internal/projectinstance/store_pg_activate.go b/internal/projectinstance/store_pg_activate.go new file mode 100644 index 00000000..1887a54d --- /dev/null +++ b/internal/projectinstance/store_pg_activate.go @@ -0,0 +1,233 @@ +package projectinstance + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "chat-agent-system/internal/workflowdef" +) + +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 +} diff --git a/internal/projectinstance/store_pg_live.go b/internal/projectinstance/store_pg_live.go new file mode 100644 index 00000000..4f8f6431 --- /dev/null +++ b/internal/projectinstance/store_pg_live.go @@ -0,0 +1,248 @@ +package projectinstance + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "chat-agent-system/internal/taskrun" +) + +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 + } + + if err := taskrun.CancelOpenByProject(ctx, tx, projectID, now, "workflow terminated"); err != nil { + return nil, err + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return inst, nil +} diff --git a/internal/projectinstance/store_pg_meta.go b/internal/projectinstance/store_pg_meta.go new file mode 100644 index 00000000..1eca4672 --- /dev/null +++ b/internal/projectinstance/store_pg_meta.go @@ -0,0 +1,114 @@ +package projectinstance + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" +) + +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 +} diff --git a/internal/projectinstance/store_pg_plan.go b/internal/projectinstance/store_pg_plan.go new file mode 100644 index 00000000..f18b049f --- /dev/null +++ b/internal/projectinstance/store_pg_plan.go @@ -0,0 +1,227 @@ +package projectinstance + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +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 +} diff --git a/internal/projectinstance/store_pg_token.go b/internal/projectinstance/store_pg_token.go new file mode 100644 index 00000000..99c52354 --- /dev/null +++ b/internal/projectinstance/store_pg_token.go @@ -0,0 +1,60 @@ +package projectinstance + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +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 +} diff --git a/internal/projectinstance/store_seed.go b/internal/projectinstance/store_seed.go new file mode 100644 index 00000000..ce8b29b9 --- /dev/null +++ b/internal/projectinstance/store_seed.go @@ -0,0 +1,20 @@ +package projectinstance + +import ( + "fmt" +) + +// 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 +}