WeiCHEN
943becfb2a
Move mem/pg activate/live/meta/plan/token/seed paths into ≤800-line siblings without behavior change; short tests green.
95 lines
4.6 KiB
Go
95 lines
4.6 KiB
Go
package projectinstance
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
||
"chat-agent-system/internal/workflowdef"
|
||
)
|
||
|
||
// TablesReady 检查 Phase 3 表是否已迁移。
|
||
func TablesReady(ctx context.Context, pool *pgxpool.Pool) bool {
|
||
if pool == nil {
|
||
return false
|
||
}
|
||
var ok bool
|
||
err := pool.QueryRow(ctx, `
|
||
SELECT EXISTS (
|
||
SELECT 1 FROM information_schema.tables
|
||
WHERE table_schema = 'public' AND table_name = 'project_workflow_instances'
|
||
)
|
||
`).Scan(&ok)
|
||
return err == nil && ok
|
||
}
|
||
|
||
// Store 项目实例持久化接口。
|
||
type Store interface {
|
||
GetProjectMeta(ctx context.Context, projectID int64) (lifecycleStatus string, activeWorkflowKey string, selectionVersion int, planRevision int, err error)
|
||
UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error
|
||
BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error)
|
||
|
||
GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
|
||
GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
|
||
// GetGoverningInstanceForAccess 供阶段/文档 ABAC：优先 live；否则最新 completed/terminated；
|
||
// 从未激活（无行或仅 planning）返回 (nil, nil)。与 HasLiveInstance（仅 active 系）语义分离。
|
||
GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error)
|
||
HasLiveInstance(ctx context.Context, projectID int64) (bool, error)
|
||
UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error)
|
||
// SavePlanWithRevision 单事务完成 planning upsert + revision bump，
|
||
// 避免 SaveWorkflowPlan 两条独立事务间的版本号与 plan 内容错位。
|
||
SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error)
|
||
ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error)
|
||
ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error)
|
||
CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error)
|
||
TerminateInstance(ctx context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error)
|
||
|
||
SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time, policyRevision *int64) error
|
||
ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error)
|
||
|
||
GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error)
|
||
AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error)
|
||
// ReopenStage sets a completed stage back to active without changing other stages.
|
||
ReopenStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) (*InstanceDetail, error)
|
||
MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error
|
||
// UpdateLiveDocTaskDuties 更新 live 实例 config_snapshot 与 document_duty_assignments（单任务）。
|
||
UpdateLiveDocTaskDuties(ctx context.Context, projectID int64, taskKey string, plan WorkflowPlan, assignedBy int64) (*WorkflowInstance, error)
|
||
// UpdateLivePlan 更新 live 快照（职责、审批策略、文档/阶段启用），并同步实例行与职责指派。
|
||
UpdateLivePlan(ctx context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, assignedBy int64) (*WorkflowInstance, error)
|
||
}
|
||
|
||
// NewStore 构建 PG 或内存 Store。
|
||
func NewStore(pool *pgxpool.Pool) Store {
|
||
if pool != nil && TablesReady(context.Background(), pool) {
|
||
return &pgStore{pool: pool}
|
||
}
|
||
return newMemStore()
|
||
}
|
||
|
||
type pgStore struct {
|
||
pool *pgxpool.Pool
|
||
}
|
||
|
||
func scanInstance(row pgx.Row) (*WorkflowInstance, error) {
|
||
var inst WorkflowInstance
|
||
var snapRaw []byte
|
||
var activatedBy, createdBy *int64
|
||
err := row.Scan(
|
||
&inst.ID, &inst.ProjectID, &inst.WorkflowVersionID, &inst.InstanceNo, &inst.Status,
|
||
&inst.ConfigRevision, &inst.PlanHash, &snapRaw, &activatedBy, &inst.ActivatedAt,
|
||
&createdBy, &inst.CreatedAt, &inst.UpdatedAt,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
inst.ActivatedBy = activatedBy
|
||
inst.CreatedBy = createdBy
|
||
if len(snapRaw) > 0 {
|
||
_ = json.Unmarshal(snapRaw, &inst.ConfigSnapshot)
|
||
}
|
||
return &inst, nil
|
||
}
|