629 lines
20 KiB
Go
629 lines
20 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/agent"
|
|
"chat-agent-system/internal/app/chatflow"
|
|
"chat-agent-system/internal/app/docflow"
|
|
"chat-agent-system/internal/app/docnotify"
|
|
"chat-agent-system/internal/app/intentflow"
|
|
"chat-agent-system/internal/app/stageflow"
|
|
"chat-agent-system/internal/app/uow"
|
|
"chat-agent-system/internal/artifactvault"
|
|
"chat-agent-system/internal/auth"
|
|
"chat-agent-system/internal/authz"
|
|
"chat-agent-system/internal/bizdev"
|
|
"chat-agent-system/internal/chat"
|
|
"chat-agent-system/internal/config"
|
|
"chat-agent-system/internal/coordinator"
|
|
"chat-agent-system/internal/ctxmgr"
|
|
"chat-agent-system/internal/docgov"
|
|
"chat-agent-system/internal/docrevision"
|
|
"chat-agent-system/internal/errcode"
|
|
"chat-agent-system/internal/errorevents"
|
|
"chat-agent-system/internal/events"
|
|
"chat-agent-system/internal/freedoc"
|
|
"chat-agent-system/internal/intent"
|
|
"chat-agent-system/internal/jevdecide"
|
|
"chat-agent-system/internal/kb"
|
|
"chat-agent-system/internal/maintenance"
|
|
"chat-agent-system/internal/notification"
|
|
"chat-agent-system/internal/observability"
|
|
"chat-agent-system/internal/projectinstance"
|
|
"chat-agent-system/internal/projectworkflow"
|
|
"chat-agent-system/internal/storage"
|
|
"chat-agent-system/internal/supplementimpact"
|
|
"chat-agent-system/internal/swboot"
|
|
"chat-agent-system/internal/sysadmin"
|
|
"chat-agent-system/internal/taskrun"
|
|
"chat-agent-system/internal/traceimpact"
|
|
"chat-agent-system/internal/workflowdef"
|
|
"chat-agent-system/internal/workflowscope"
|
|
"chat-agent-system/internal/workspace"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
const defaultReminderPollInterval = 15 * time.Minute
|
|
|
|
// Runtime is the unique composition root. cmd/server, tests and workers
|
|
// construct it once; HTTP adapters must not re-wire domain services.
|
|
type Runtime struct {
|
|
Cfg *config.Config
|
|
Biz *bizdev.Service
|
|
Engine *coordinator.CoordinatorEngine
|
|
Auth *auth.Service
|
|
PW *projectworkflow.Service
|
|
TaskRuns *taskrun.Manager
|
|
Authorizer *authz.Authorizer
|
|
Authz *authz.Service
|
|
WFDef *workflowdef.Service
|
|
Inst *projectinstance.Service
|
|
Notify *notification.Service
|
|
Resolver *workspace.Resolver
|
|
WSCap *workspace.Capability
|
|
Commands *taskrun.Commands
|
|
Rec intent.Recognizer
|
|
Stageflow *stageflow.UseCase
|
|
DocNotify *docnotify.Subscriber
|
|
ChatLog chatflow.ChatLog
|
|
Assembler chatflow.ContextAssembler
|
|
Docflow *docflow.Commands
|
|
IntentPlans *intentflow.Plans
|
|
Pool *pgxpool.Pool
|
|
Gov *docgov.GovernanceEngine
|
|
Rev *docrevision.Service
|
|
Trace *traceimpact.Service
|
|
SupplementImpact *supplementimpact.Service
|
|
mutateOwner *MutateOwner
|
|
ChatReopens chat.ReopenService
|
|
Ctx *ctxmgr.ContextManager
|
|
CAS *storage.CAS
|
|
FreeDoc *freedoc.Service
|
|
KB *kb.Client
|
|
SysAdmin *sysadmin.Service
|
|
Maintenance *maintenance.Service
|
|
|
|
mu sync.Mutex
|
|
running bool
|
|
subscribers int
|
|
starts int
|
|
notifyBound bool
|
|
hookBound bool
|
|
reminderCancel context.CancelFunc
|
|
}
|
|
|
|
func New(cfg *config.Config) (*Runtime, error) {
|
|
return Assemble(cfg, nil)
|
|
}
|
|
|
|
// Assemble wires domain services. svc may be injected by tests.
|
|
func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
|
|
if cfg == nil {
|
|
cfg, _ = config.Load("")
|
|
}
|
|
if cfg != nil {
|
|
observability.InitLogger(cfg.Env)
|
|
}
|
|
// FIND-QUAL-01：组合根启动期校验双套 flag 一致性（fail-fast）。
|
|
if err := intentflow.ValidateFlagConsistency(intent.CurrentFeatureFlags(), intentflow.CurrentFeatureFlags()); err != nil {
|
|
return nil, err
|
|
}
|
|
if svc == nil {
|
|
var err error
|
|
svc, err = bizdev.NewService(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
rec := intent.NewRecognizer()
|
|
backendName := ""
|
|
if cfg != nil {
|
|
backendName = cfg.AgentBackend
|
|
}
|
|
dialogueBackend := agent.NewBackend(backendName, cfg)
|
|
coordEngine := coordinator.NewCoordinatorEngineWithDialogue(nil, dialogueBackend, cfg.DataDir)
|
|
if cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != "" {
|
|
coordEngine.SetRealtimeSearcher(&agent.PiRunnerSearcher{
|
|
URL: cfg.PiAgentRunnerURL,
|
|
Token: cfg.PiAgentRunnerAuthToken,
|
|
})
|
|
}
|
|
// jev-service typed decide: default off; shadow/assist inject client. Fail-open on errors.
|
|
if jevdecide.Enabled() {
|
|
endpoint, key := "", ""
|
|
if cfg != nil {
|
|
endpoint = strings.TrimSpace(cfg.JevBaseURL)
|
|
key = strings.TrimSpace(cfg.JevAPIKey)
|
|
}
|
|
var jc *jevdecide.Client
|
|
if endpoint != "" || key != "" {
|
|
jc = jevdecide.NewClient(endpoint, key)
|
|
if endpoint == "" {
|
|
jc = jevdecide.NewClientFromEnv()
|
|
}
|
|
} else {
|
|
jc = jevdecide.NewClientFromEnv()
|
|
}
|
|
if jc != nil && jc.Endpoint != "" {
|
|
coordEngine.SetDecideAssist(jc)
|
|
slog.Info("jevdecide wired", "mode", string(jevdecide.CurrentMode()), "endpoint", jc.Endpoint)
|
|
}
|
|
}
|
|
parts := bizdev.Snapshot(svc)
|
|
if ctxmgr.SemanticSummaryShadowEnabled() && parts.Ctx != nil && backendName == "pi-agent" {
|
|
model := ""
|
|
dataDir := ""
|
|
if cfg != nil {
|
|
model = cfg.PiAgentModel
|
|
dataDir = cfg.DataDir
|
|
}
|
|
parts.Ctx.SetSemanticSummarizer(&PiSemanticSummarizer{Backend: dialogueBackend, Model: model, DataDir: dataDir})
|
|
}
|
|
if docrevision.PIChangeReportNarrativeEnabled() && parts.Rev != nil && backendName == "pi-agent" {
|
|
model, dataDir := "", ""
|
|
if cfg != nil {
|
|
model = cfg.PiAgentModel
|
|
dataDir = cfg.DataDir
|
|
}
|
|
narr := &PiChangeReportNarrator{Backend: dialogueBackend, Model: model, DataDir: dataDir}
|
|
parts.Rev.NarrativeFetcher = narr.Fetch
|
|
}
|
|
if intentflow.CurrentFeatureFlags().U9Segment && backendName == "pi-agent" {
|
|
model, dataDir := "", ""
|
|
if cfg != nil {
|
|
model = cfg.PiAgentModel
|
|
dataDir = cfg.DataDir
|
|
}
|
|
intentflow.SetLLMSegmentProposer(&PiSegmentProposer{Backend: dialogueBackend, Model: model, DataDir: dataDir})
|
|
}
|
|
pool := parts.Pool
|
|
if pool != nil {
|
|
errorevents.Install(errorevents.NewStore(pool))
|
|
}
|
|
stores, storeErr := auth.NewStores(context.Background(), pool, cfg)
|
|
if storeErr != nil {
|
|
return nil, fmt.Errorf("auth store: %w", storeErr)
|
|
}
|
|
authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg)
|
|
pwSvc := parts.PW
|
|
if pwSvc == nil {
|
|
pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
|
|
}
|
|
|
|
authzSvc := authz.NewService(pool)
|
|
wfDefSvc := workflowdef.NewService(pool, parts.CAS)
|
|
instStore := projectinstance.NewStore(pool)
|
|
var instSvc *projectinstance.Service
|
|
if wfDefSvc.Available() {
|
|
docChecker := projectinstance.NewDocGovChecker(parts.Gov)
|
|
instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
|
|
if pool != nil {
|
|
instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool))
|
|
}
|
|
if parts.Gov != nil {
|
|
instSvc.SetDocumentDutyProjector(&documentDutyProjectorAdapter{gov: parts.Gov})
|
|
instSvc.SetTemplateDocMaterializer(&templateDocMaterializerAdapter{gov: parts.Gov})
|
|
}
|
|
}
|
|
taskMgr := taskrun.NewManager(pwSvc, parts.Ctx)
|
|
if pool != nil {
|
|
taskMgr.SetStore(taskrun.NewPGStore(pool))
|
|
}
|
|
if instSvc != nil && parts.Gov != nil {
|
|
instSvc.SetDocTaskBusyChecker(&docTaskBusyCheckerAdapter{gov: parts.Gov, taskRuns: taskMgr})
|
|
}
|
|
wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
|
|
// SWBOOT-10: unlock controlled_patch creation only for admin-allowlisted
|
|
// repo bindings (repo_path_allowlist); nil pool keeps fail-closed rejection.
|
|
if pool != nil {
|
|
if al := swboot.NewPGRepoAllowlist(pool, 30*time.Second); al != nil {
|
|
pwSvc.RepoAllowlistChecker = al.IsAllowlisted
|
|
}
|
|
}
|
|
cmds := &taskrun.Commands{
|
|
Ctx: parts.Ctx,
|
|
PW: pwSvc,
|
|
Env: cfg.Env,
|
|
StorageIdentity: "env:" + cfg.Env,
|
|
}
|
|
if pool != nil {
|
|
cmds.UoW = uow.NewPG(pool)
|
|
cmds.Fav = &artifactvault.Ingest{Env: cfg.Env}
|
|
}
|
|
taskMgr.SetCommands(cmds)
|
|
|
|
stages := stageflow.New(svc)
|
|
coordEngine.SetStageTaskRunner(stages)
|
|
coordEngine.SetContextServices(parts.Ctx, parts.KB)
|
|
coordEngine.SetSessionRunReader(sessionRunAdapter{mgr: taskMgr})
|
|
if instSvc != nil && instSvc.Available() {
|
|
coordEngine.SetInstanceService(instSvc)
|
|
}
|
|
traceSvc := traceimpact.Open(pool)
|
|
traceSvc.SetStageReopener(traceStage{inst: instSvc})
|
|
traceSvc.SetDocUnlocker(traceUnlock{gov: parts.Gov})
|
|
bindRecheckChat(traceSvc, parts.Ctx)
|
|
if instSvc != nil {
|
|
instSvc.SetRecheckGate(traceRecheck{svc: traceSvc})
|
|
}
|
|
coordEngine.SetImpactActions(traceChat{svc: traceSvc})
|
|
if pool != nil {
|
|
access := authz.NewAuthorizer(pool)
|
|
wireDutySlotWriter(access)
|
|
coordEngine.SetProjectAccess(func(ctx context.Context, projectID, userID int64) error {
|
|
return access.RequireProjectAccess(ctx, projectID, userID)
|
|
})
|
|
}
|
|
if parts.Rev != nil {
|
|
wireDocRevisionPorts(parts.Rev)
|
|
parts.Rev.CoverageResolver = func(ctx context.Context, docID string) string {
|
|
projectID := projectIDFromControlledDoc(docID)
|
|
if projectID == 0 {
|
|
return traceimpact.ImpactIndexPending
|
|
}
|
|
return traceSvc.CoverageForDoc(ctx, projectID, docID)
|
|
}
|
|
}
|
|
|
|
authorizer := authz.NewAuthorizer(pool)
|
|
if pool == nil && cfg != nil && cfg.AllowInMemory {
|
|
// 显式 ALLOW_IN_MEMORY 演示模式：无库时采用内存授权语义；
|
|
// 误配（无库且未显式声明）仍走 NewAuthorizer(nil) 的严格拒绝。
|
|
authorizer = authz.NewInMemoryAuthorizer()
|
|
}
|
|
wireDutySlotWriter(authorizer)
|
|
rt := &Runtime{
|
|
Cfg: cfg,
|
|
Biz: svc,
|
|
Engine: coordEngine,
|
|
Auth: authSvc,
|
|
PW: pwSvc,
|
|
TaskRuns: taskMgr,
|
|
Authorizer: authorizer,
|
|
Authz: authzSvc,
|
|
WFDef: wfDefSvc,
|
|
Inst: instSvc,
|
|
Notify: notification.NewService(pool),
|
|
Resolver: workspace.NewResolver(cfg.DataDir, pool),
|
|
WSCap: wsCap,
|
|
Commands: cmds,
|
|
Rec: rec,
|
|
Stageflow: stages,
|
|
Pool: pool,
|
|
Gov: parts.Gov,
|
|
Rev: parts.Rev,
|
|
Trace: traceSvc,
|
|
Ctx: parts.Ctx,
|
|
CAS: parts.CAS,
|
|
FreeDoc: parts.FreeDoc,
|
|
KB: parts.KB,
|
|
}
|
|
rt.SysAdmin = sysadmin.NewService(cfg, sysadmin.Options{
|
|
Pool: pool,
|
|
Status: svc,
|
|
KB: parts.KB,
|
|
TaskStoreReady: func(ctx context.Context) bool {
|
|
return taskrun.TablesReady(ctx, pool)
|
|
},
|
|
FeatureValues: sysadminFeatureValues(cfg),
|
|
})
|
|
if cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != "" {
|
|
rt.SysAdmin.SetApplyFn(sysadmin.NewRunnerApplyFunc(
|
|
rt.SysAdmin, cfg.PiAgentRunnerURL, cfg.PiAgentRunnerAuthToken, nil,
|
|
))
|
|
go func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
if _, err := rt.SysAdmin.ApplyToRunner(ctx); err != nil {
|
|
fmt.Printf("sysadmin: initial runner apply: %v\n", err)
|
|
}
|
|
}()
|
|
}
|
|
if pool != nil {
|
|
rt.Maintenance = maintenance.NewService(maintenance.NewPGStore(pool))
|
|
if rt.SysAdmin != nil {
|
|
// P9 白名单消费方接线：ui.maintenance_notice / ui.read_only_mode
|
|
// 经公开维护状态端点生效（横幅常驻 + 前端只读提示）。
|
|
sa := rt.SysAdmin
|
|
rt.Maintenance.SetSiteSettings(func(ctx context.Context) (string, bool) {
|
|
notice, _ := sa.GetSettingValue(ctx, "ui.maintenance_notice")
|
|
noticeStr, _ := notice.(string)
|
|
readOnly := false
|
|
if v, err := sa.GetSettingValue(ctx, "ui.read_only_mode"); err == nil {
|
|
if b, ok := v.(bool); ok {
|
|
readOnly = b
|
|
}
|
|
}
|
|
return noticeStr, readOnly
|
|
})
|
|
}
|
|
}
|
|
if instSvc != nil {
|
|
instSvc.SetAuthorOnlyPolicySource(&sysadminAuthorOnlyPolicyAdapter{svc: rt.SysAdmin})
|
|
svc.SetInstanceDutySource(&projectInstanceDutyAdapter{inst: instSvc})
|
|
}
|
|
rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: parts.Gov, Chat: parts.Ctx}
|
|
rt.DocNotify.SkipWrite = func(ctx context.Context) bool {
|
|
return rt.WriterLive(ctx, events.ConsumerNotify)
|
|
}
|
|
rt.ChatReopens = rt.WireDocDialogueTriggers()
|
|
envName := ""
|
|
if cfg != nil {
|
|
envName = cfg.Env
|
|
}
|
|
rt.Docflow = &docflow.Commands{
|
|
Gov: parts.Gov, Rev: parts.Rev, Auth: docflow.DutyAuthorizer(),
|
|
Environment: envName,
|
|
}
|
|
if pool != nil {
|
|
rt.Docflow.UoW = uow.NewPG(pool)
|
|
rt.Docflow.Events = events.PGSink{}
|
|
livePI := cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != ""
|
|
taskOwner := NewTaskRunAdapter()
|
|
taskOwner.Starter = LiveTaskStarter(taskMgr, livePI)
|
|
queryOwner := NewQueryOwner(projectAccess(rt), parts.KB)
|
|
pflags := intentflow.CurrentFeatureFlags()
|
|
liveOwners := intentflow.LiveOwners{}
|
|
if pflags.U10Mutate {
|
|
rt.mutateOwner = NewMutateOwner(projectAccess(rt), projectBlackboardPort(pool, parts.Ctx))
|
|
liveOwners.Mutate = rt.mutateOwner
|
|
}
|
|
if pflags.U11DocGov {
|
|
liveOwners.DocGov = NewDocGovOwner(parts.Gov, projectAccess(rt), nil)
|
|
}
|
|
if pflags.U12RunControl {
|
|
rc := NewRunControlOwner(projectAccess(rt), pwSvc, taskMgr)
|
|
rc.ProjectName = projectNameLookup(pool)
|
|
liveOwners.RunControl = rc
|
|
}
|
|
rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{
|
|
Owner: intentflow.BuildOwnerMuxWithLive(
|
|
intentflow.QueryGuard{Inner: queryOwner},
|
|
taskOwner,
|
|
liveOwners,
|
|
),
|
|
Events: events.PGSink{},
|
|
ResourceVersionBinder: newResourceVersionBinder(projectBlackboardPort(pool, parts.Ctx), parts.Gov),
|
|
})
|
|
}
|
|
svc.SetAgentMaterializer(rt.Docflow)
|
|
rt.bindDocumentNotify()
|
|
rt.bindWorkflowApprovalHook()
|
|
rt.wireSupplementImpact()
|
|
return rt, nil
|
|
}
|
|
|
|
func (r *Runtime) bindDocumentNotify() {
|
|
if r == nil || r.notifyBound {
|
|
return
|
|
}
|
|
// Durable notify is live; do not assign OnOutbox in production.
|
|
r.notifyBound = true
|
|
}
|
|
|
|
func (r *Runtime) bindWorkflowApprovalHook() {
|
|
if r == nil || r.hookBound {
|
|
return
|
|
}
|
|
// Durable workflow_advance is live; do not register AddOnApprovedHook in production.
|
|
r.hookBound = true
|
|
}
|
|
|
|
// sysadminFeatureValues 在组合根解析各业务包公开 flag 的生效值，供 sysadmin 聚合层
|
|
// 只读消费；sysadmin 不 import 业务包。
|
|
func sysadminFeatureValues(cfg *config.Config) func() map[string]string {
|
|
return func() map[string]string {
|
|
iflags := intent.CurrentFeatureFlags()
|
|
pflags := intentflow.CurrentFeatureFlags()
|
|
sflags := ctxmgr.CurrentSummaryFlags()
|
|
values := map[string]string{
|
|
"INTENT_V2_SHADOW": strconv.FormatBool(iflags.IntentV2Shadow),
|
|
"INTENT_V2_DECISION": strconv.FormatBool(iflags.IntentV2Decision),
|
|
"INTENT_V2_DECISION_TIER": string(iflags.V2DecisionTier),
|
|
"WORKFLOW_INTENTS_ENABLED": strconv.FormatBool(iflags.WorkflowIntentsEnabled),
|
|
"MULTI_INTENT_ENABLED": strconv.FormatBool(iflags.MultiIntentEnabled),
|
|
"MULTI_INTENT_PLAN_TIER": string(pflags.PlanTier),
|
|
"INTENT_PLAN_PROJECT_ALLOWLIST": formatProjectAllowlist(pflags.ProjectAllowlist),
|
|
"MULTI_INTENT_U9_SEGMENT": strconv.FormatBool(pflags.U9Segment),
|
|
"MULTI_INTENT_U10_MUTATE": strconv.FormatBool(pflags.U10Mutate),
|
|
"MULTI_INTENT_U11_DOCGOV": strconv.FormatBool(pflags.U11DocGov),
|
|
"MULTI_INTENT_U12_RUN_CONTROL": strconv.FormatBool(pflags.U12RunControl),
|
|
"WORKFLOW_SCOPE_MODE": string(workflowscope.CurrentMode()),
|
|
"SEMANTIC_SUMMARY_SHADOW": strconv.FormatBool(sflags.Shadow),
|
|
"SEMANTIC_SUMMARY_REPLACE": strconv.FormatBool(sflags.Replace),
|
|
"PI_CHANGE_REPORT_NARRATIVE": strconv.FormatBool(docrevision.PIChangeReportNarrativeEnabled()),
|
|
}
|
|
if cfg != nil {
|
|
values["AGENT_BACKEND"] = cfg.AgentBackend
|
|
values["MAX_CONCURRENT_AGENTS"] = strconv.Itoa(cfg.MaxConcurrentAgents)
|
|
values["PI_AGENT_RUNNER_URL"] = cfg.PiAgentRunnerURL
|
|
values["ALLOW_IN_MEMORY"] = strconv.FormatBool(cfg.AllowInMemory)
|
|
values["AUTH_REGISTRATION_ENABLED"] = strconv.FormatBool(cfg.AuthRegistrationEnabled)
|
|
values["AUTH_REQUIRE_APPROVAL"] = strconv.FormatBool(cfg.AuthRequireApproval)
|
|
values["FAV_PUBLISHER_ENABLED"] = strconv.FormatBool(cfg.FAVPublisherEnabled)
|
|
values["FAV_DEGRADED_READ"] = strconv.FormatBool(cfg.FAVDegradedRead)
|
|
}
|
|
return values
|
|
}
|
|
}
|
|
|
|
func projectAccess(rt *Runtime) ProjectAccess {
|
|
return func(ctx context.Context, projectID, userID int64) error {
|
|
if rt == nil || rt.Authorizer == nil {
|
|
return intentflow.ErrPlanActionForbidden
|
|
}
|
|
return rt.Authorizer.RequireProjectAccess(ctx, projectID, userID)
|
|
}
|
|
}
|
|
|
|
func formatProjectAllowlist(ids []int64) string {
|
|
if len(ids) == 0 {
|
|
return ""
|
|
}
|
|
parts := make([]string, 0, len(ids))
|
|
for _, id := range ids {
|
|
if id == intentflow.AllowAllProjects {
|
|
return "*"
|
|
}
|
|
parts = append(parts, strconv.FormatInt(id, 10))
|
|
}
|
|
return strings.Join(parts, ",")
|
|
}
|
|
|
|
// WriterLive reports whether the durable consumer already owns the visible write.
|
|
func (r *Runtime) WriterLive(ctx context.Context, consumer string) bool {
|
|
if r == nil {
|
|
return false
|
|
}
|
|
env := ""
|
|
if r.Cfg != nil {
|
|
env = r.Cfg.Env
|
|
}
|
|
return events.WriterLive(ctx, r.Pool, env, consumer)
|
|
}
|
|
|
|
func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) {
|
|
if r.WriterLive(ctx, events.ConsumerWorkflowAdvance) {
|
|
return
|
|
}
|
|
if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil {
|
|
return
|
|
}
|
|
result, err := r.Inst.HandleDocumentApproved(ctx, doc.ProjectID, doc.TaskKey)
|
|
if err != nil || result == nil || !result.StageReady {
|
|
return
|
|
}
|
|
if r.Notify == nil || !r.Notify.Available() {
|
|
return
|
|
}
|
|
if r.PW == nil {
|
|
return
|
|
}
|
|
project, err := r.PW.GetProject(ctx, doc.ProjectID, 0)
|
|
if err != nil || project == nil || project.CreatorUserID <= 0 {
|
|
return
|
|
}
|
|
_, _ = r.Notify.EmitStageReadyToAdvance(ctx, project.CreatorUserID, notification.StageReadyPayload{
|
|
ProjectID: doc.ProjectID,
|
|
WorkflowInstanceID: result.InstanceID,
|
|
StageInstanceID: result.StageInstanceID,
|
|
StageKey: result.StageKey,
|
|
StageName: result.StageName,
|
|
ConfigRevision: result.ConfigRevision,
|
|
RequiredDocCount: result.RequiredDocCount,
|
|
ApprovedDocCount: result.ApprovedDocCount,
|
|
})
|
|
}
|
|
|
|
// Start launches background workers. Repeat calls on a running runtime are no-ops.
|
|
func (r *Runtime) Start(_ context.Context) error {
|
|
if r == nil {
|
|
return fmt.Errorf("app: nil runtime")
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.running {
|
|
return nil
|
|
}
|
|
r.bindDocumentNotify()
|
|
r.bindWorkflowApprovalHook()
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
r.reminderCancel = cancel
|
|
go r.runReminderLoop(ctx)
|
|
r.running = true
|
|
r.starts++
|
|
if r.subscribers == 0 {
|
|
r.subscribers = 1
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Stop cancels workers. A later Start recreates the reminder loop without
|
|
// double-registering document subscribers.
|
|
func (r *Runtime) Stop(_ context.Context) error {
|
|
if r == nil {
|
|
return nil
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if !r.running {
|
|
return nil
|
|
}
|
|
if r.reminderCancel != nil {
|
|
r.reminderCancel()
|
|
r.reminderCancel = nil
|
|
}
|
|
r.running = false
|
|
return nil
|
|
}
|
|
|
|
func (r *Runtime) SubscriberCount() int {
|
|
if r == nil {
|
|
return 0
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.subscribers
|
|
}
|
|
|
|
func (r *Runtime) StartCount() int {
|
|
if r == nil {
|
|
return 0
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.starts
|
|
}
|
|
|
|
func (r *Runtime) Running() bool {
|
|
if r == nil {
|
|
return false
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.running
|
|
}
|
|
|
|
func (r *Runtime) runReminderLoop(ctx context.Context) {
|
|
r.runStageAdvanceReminders(ctx)
|
|
ticker := time.NewTicker(defaultReminderPollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
r.runStageAdvanceReminders(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *Runtime) runStageAdvanceReminders(ctx context.Context) {
|
|
if r.Notify == nil || !r.Notify.Available() {
|
|
return
|
|
}
|
|
n, err := r.Notify.ProcessStageAdvanceReminders(ctx)
|
|
if err != nil {
|
|
slog.Warn("stage advance reminder scan failed", "error", err)
|
|
errcode.Report(ctx, errcode.Event{
|
|
Code: "REMINDER_SCAN_FAILED",
|
|
Module: "notification",
|
|
Source: "async",
|
|
Upstream: err.Error(),
|
|
})
|
|
return
|
|
}
|
|
if n > 0 {
|
|
slog.Info("stage advance 24h reminders sent", "count", n)
|
|
}
|
|
}
|