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