Agent-System/internal/events/cutover.go

package events

import (
	"context"
	"fmt"
	"strings"

	"chat-agent-system/internal/app/uow"

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

const OpsCutoverEnvironment = "dev"

// ValidateOpsCutover is the shared-dev gate for cluster mode changes.
// test/main, FAV, and unknown consumers are rejected before any write.
func ValidateOpsCutover(environment, consumer, mode string) error {
	if strings.TrimSpace(environment) != OpsCutoverEnvironment {
		return fmt.Errorf("%w: environment %q", ErrOpsCutover, environment)
	}
	switch consumer {
	case ConsumerNotify, ConsumerWorkflowAdvance, ConsumerProjection:
	default:
		return fmt.Errorf("%w: consumer %q", ErrOpsCutover, consumer)
	}
	switch mode {
	case ModeOff, ModeShadow, ModeLive:
	default:
		return fmt.Errorf("%w: mode %q", ErrOpsCutover, mode)
	}
	return nil
}

func ApplyOpsCutover(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error {
	if err := ValidateOpsCutover(environment, consumer, mode); err != nil {
		return err
	}
	return SetMode(ctx, pool, environment, consumer, mode)
}

type ModeSnapshot struct {
	Environment string `json:"environment"`
	Consumer    string `json:"consumer"`
	Mode        string `json:"mode"`
	Revision    int64  `json:"revision"`
}

type BacklogRow struct {
	Consumer string `json:"consumer"`
	Status   string `json:"status"`
	Count    int64  `json:"count"`
}

func ListModes(ctx context.Context, pool *pgxpool.Pool) ([]ModeSnapshot, error) {
	rows, err := pool.Query(ctx, `
		SELECT environment, consumer, mode, revision
		FROM event_consumer_modes
		ORDER BY environment, consumer
	`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ModeSnapshot
	for rows.Next() {
		var m ModeSnapshot
		if err := rows.Scan(&m.Environment, &m.Consumer, &m.Mode, &m.Revision); err != nil {
			return nil, err
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

func ListBacklog(ctx context.Context, pool *pgxpool.Pool) ([]BacklogRow, error) {
	rows, err := pool.Query(ctx, `
		SELECT consumer, status, COUNT(*)
		FROM event_deliveries
		GROUP BY consumer, status
		ORDER BY consumer, status
	`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []BacklogRow
	for rows.Next() {
		var r BacklogRow
		if err := rows.Scan(&r.Consumer, &r.Status, &r.Count); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

type pendingDelivery struct {
	EventID       string
	Consumer      string
	EventName     string
	Environment   string
	ProjectID     int64
	SchemaVersion int
	PayloadDigest string
}

func loadPending(ctx context.Context, pool *pgxpool.Pool) ([]pendingDelivery, error) {
	rows, err := pool.Query(ctx, `
		SELECT d.event_id, d.consumer, e.event_name, e.environment, e.project_id, e.schema_version, COALESCE(e.payload_digest, '')
		FROM event_deliveries d
		INNER JOIN event_envelopes e ON e.event_id = d.event_id
		WHERE d.status IN ('pending','leased')
	`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []pendingDelivery
	for rows.Next() {
		var p pendingDelivery
		if err := rows.Scan(&p.EventID, &p.Consumer, &p.EventName, &p.Environment, &p.ProjectID, &p.SchemaVersion, &p.PayloadDigest); err != nil {
			return nil, err
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

func ClassifyPending(ctx context.Context, pool *pgxpool.Pool) ([]InventoryRow, error) {
	pending, err := loadPending(ctx, pool)
	if err != nil {
		return nil, err
	}
	out := make([]InventoryRow, 0, len(pending))
	for _, p := range pending {
		env := Envelope{
			EventID: p.EventID, EventName: p.EventName, SchemaVersion: p.SchemaVersion,
			Environment: p.Environment, ProjectID: p.ProjectID, PayloadDigest: p.PayloadDigest,
		}
		out = append(out, ClassifyEnvelope(env, p.Consumer))
	}
	return out, nil
}

// QuarantineIllegalPending persists quarantine for pending deliveries whose
// consumer is not on the contract (including FAV). Safe envelopes are left untouched.
func QuarantineIllegalPending(ctx context.Context, pool *pgxpool.Pool, operator, approval string) (int, error) {
	if operator == "" || approval == "" {
		return 0, ErrIntegrityFailure
	}
	rows, err := ClassifyPending(ctx, pool)
	if err != nil {
		return 0, err
	}
	n := 0
	unit := uow.NewPG(pool)
	for _, row := range rows {
		if row.Classification != ClassificationQuarantine {
			continue
		}
		err := unit.Within(ctx, func(scope uow.TxScope) error {
			return PersistInventory(uow.WithScope(ctx, scope), row, operator, approval)
		})
		if err != nil {
			return n, err
		}
		n++
	}
	return n, nil
}