refactor(projectinstance): split store.go into use-case files
Move mem/pg activate/live/meta/plan/token/seed paths into ≤800-line siblings without behavior change; short tests green.
This commit is contained in:
WeiCHEN
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user