Agent-System/internal/app/eventsconsumer/eventsconsumer.go

package eventsconsumer

import (
	"context"
	"fmt"
	"os"
	"time"

	"chat-agent-system/internal/app/docadvance"
	"chat-agent-system/internal/app/docnotify"
	"chat-agent-system/internal/app/docproject"
	"chat-agent-system/internal/app/intentflow"
	"chat-agent-system/internal/config"
	"chat-agent-system/internal/ctxmgr"
	"chat-agent-system/internal/docgov"
	"chat-agent-system/internal/events"
	"chat-agent-system/internal/notification"
	"chat-agent-system/internal/projectinstance"
	"chat-agent-system/internal/projectworkflow"
	"chat-agent-system/internal/workflowdef"

	"github.com/jackc/pgx/v5/pgxpool"
)

// Handle owns an in-process events dispatcher loop.
type Handle struct {
	Disp   *events.Dispatcher
	Owner  string
	cancel context.CancelFunc
}

// DefaultConsumers is the production consumer set (FAV intentionally omitted).
func DefaultConsumers() []string {
	return []string{
		events.ConsumerNotify,
		events.ConsumerProjection,
		events.ConsumerWorkflowAdvance,
	}
}

// Enabled reports whether in-process events should run inside cmd/server.
// Default true; set EVENTS_INPROCESS=0|false|off to disable (use standalone cmd/events).
func Enabled() bool {
	v := os.Getenv("EVENTS_INPROCESS")
	switch v {
	case "0", "false", "off", "no":
		return false
	default:
		return true
	}
}

// Start registers consumers and runs Dispatcher.Serve in a goroutine until ctx cancels.
func Start(parent context.Context, pool *pgxpool.Pool, cfg *config.Config) (*Handle, error) {
	if pool == nil {
		return nil, fmt.Errorf("eventsconsumer: pool required")
	}
	if cfg == nil {
		return nil, fmt.Errorf("eventsconsumer: config required")
	}
	owner := fmt.Sprintf("events-%d", os.Getpid())
	disp := events.NewDispatcher(pool, cfg.Env, owner)
	if err := registerConsumers(disp, pool, cfg); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(parent)
	go func() {
		_ = disp.Serve(ctx, DefaultConsumers())
	}()
	return &Handle{Disp: disp, Owner: owner, cancel: cancel}, nil
}

func registerConsumers(disp *events.Dispatcher, pool *pgxpool.Pool, cfg *config.Config) error {
	notify := notification.NewService(pool)
	gov := docgov.NewGovernanceEngine(pool)
	wfDef := workflowdef.NewService(pool, nil)
	pwSvc := projectworkflow.NewService(projectworkflow.NewStore(pool), nil)
	instSvc := projectinstance.NewService(
		projectinstance.NewStore(pool), wfDef.Store(), pwSvc.Store(), projectinstance.NewDocGovChecker(gov),
	)
	if err := disp.Register(events.ConsumerNotify, (&docnotify.EventConsumer{
		Notify: notify, Docs: gov, Env: cfg.Env,
		Chat: ctxmgr.NewContextManager(pool, nil, ctxmgr.TimeDecayConfig{}),
	}).HandleLease); err != nil {
		return err
	}
	if err := disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{
		Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env,
	}).HandleLease); err != nil {
		return err
	}
	if err := disp.Register(events.ConsumerProjection, intentflow.MuxProjection(
		(&docproject.EventConsumer{Docs: gov, Env: cfg.Env}).HandleLease,
		(&intentflow.ProjectionConsumer{Env: cfg.Env}).HandleLease,
	)); err != nil {
		return err
	}
	// Intentionally NOT registering events.ConsumerFAV / internal/app/fav.
	return nil
}

// Stop requests claim stop and cancels the serve loop.
func (h *Handle) Stop() {
	if h == nil {
		return
	}
	if h.Disp != nil {
		h.Disp.StopClaim()
	}
	if h.cancel != nil {
		h.cancel()
	}
}

// Alive reports whether the dispatcher heartbeat is fresh.
func (h *Handle) Alive(maxAge time.Duration) bool {
	if h == nil || h.Disp == nil {
		return false
	}
	return h.Disp.Alive(maxAge)
}