Agent-System/internal/events/envelope.go

package events

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"strings"
	"time"
)

const EnvelopeSchema = 1

var (
	ErrQuarantine       = errors.New("events: envelope quarantined")
	ErrStaleFencing     = errors.New("events: stale delivery fencing token")
	ErrDoubleOwner      = errors.New("events: delivery already leased")
	ErrGapWait          = errors.New("events: aggregate version gap")
	ErrAuthFailure      = errors.New("events: authorization failure")
	ErrIntegrityFailure = errors.New("events: integrity failure")
	ErrReplayScope      = errors.New("events: replay scope rejected")
	ErrUnknownConsumer  = errors.New("events: unknown consumer")
	ErrInvalidScope     = errors.New("events: environment or project scope missing")
	ErrDisabledConsumer = errors.New("events: consumer is disabled")
	ErrUnregistered     = errors.New("events: event or consumer is not in the static registry")
	ErrReplayApply      = errors.New("events: replay apply rejected")
	ErrOpsCutover       = errors.New("events: shared cutover rejected")
)

type Envelope struct {
	EventID          string
	EventName        string
	SchemaVersion    int
	Environment      string
	ProjectID        int64
	AggregateType    string
	AggregateID      string
	AggregateVersion int64
	ActorRef         string
	CorrelationID    string
	CausationID      string
	IdempotencyKey   string
	PayloadDigest    string
	Producer         string
	OccurredAt       time.Time
	Payload          []byte
}

type Delivery struct {
	EventID       string
	Consumer      string
	Status        string // pending, leased, acked, dead, quarantine
	Attempt       int
	LeaseOwner    string
	FencingToken  string
	LeaseUntil    time.Time
	NextAttemptAt time.Time
	Inbox         bool
}

func Digest(payload []byte) string {
	sum := sha256.Sum256(payload)
	return hex.EncodeToString(sum[:])
}

func ValidateEnvelope(env Envelope) error {
	if env.EventID == "" || env.EventName == "" || env.AggregateID == "" {
		return ErrIntegrityFailure
	}
	if env.SchemaVersion != EnvelopeSchema {
		return ErrIntegrityFailure
	}
	if strings.Contains(strings.ToLower(string(env.Payload)), "authorization: bearer") ||
		strings.Contains(strings.ToLower(string(env.Payload)), "api_key=") {
		return ErrAuthFailure
	}
	if env.PayloadDigest == "" || env.PayloadDigest != Digest(env.Payload) {
		return ErrIntegrityFailure
	}
	if env.Producer == "" || strings.EqualFold(env.Producer, "forged") {
		return ErrIntegrityFailure
	}
	if strings.TrimSpace(env.Environment) == "" || env.ProjectID <= 0 {
		return ErrInvalidScope
	}
	return nil
}