WeiCHEN
943becfb2a
Move mem/pg activate/live/meta/plan/token/seed paths into ≤800-line siblings without behavior change; short tests green.
228 lines
6.9 KiB
Go
228 lines
6.9 KiB
Go
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
|
|
}
|