consumer

type BufferedEvent ¶

type BufferedEvent struct {
	Event        publisher.Event
	Timestamp    time.Time
	Type         publisher.EventType
	AttemptCount int
}

BufferedEvent represents an event waiting for reconciliation

type Classifier ¶

type Classifier struct {
	// contains filtered or unexported fields
}

Classifier determines if an event can be immediately processed or needs buffering

func NewClassifier(repo *repository.MongoRepository, logger *slog.Logger) *Classifier

func (c *Classifier) Classify(ctx context.Context, event publisher.Event) (EventClassification, error)

type EventClassification ¶

type EventClassification string

EventClassification indicates how an event should be processed

const (
	// ClassifyImmediate means the event can be processed immediately
	ClassifyImmediate EventClassification = "immediate"
	// ClassifyBuffer means the event should be buffered for reconciliation
	ClassifyBuffer EventClassification = "buffer"
	// ClassifyReconcile means the event is a root suite end that triggers reconciliation
	ClassifyReconcile EventClassification = "reconcile"
)

type MongoNATSConsumer ¶

type MongoNATSConsumer struct {
	// contains filtered or unexported fields
}

MongoNATSConsumer wraps a NATS JetStream consumer for processing test events and persisting them to MongoDB using a document-based data model.

Event Processing Flow: - Suite Begin (root): Creates a new TestRunDocument if it doesn't exist - Suite Begin (nested): Finds parent suite in root document and upserts the nested suite - Test Begin: Finds parent suite in root document and upserts the test - Step Begin: Finds parent test in root document and upserts the step - End Events: Find and update existing entities within the root document

All operations follow an upsert pattern: update if entity exists, insert if not found.

func NewMongoNATSConsumer(cfg MongoNATSConsumerConfig, logger *slog.Logger, repo *repository.MongoRepository) (*MongoNATSConsumer, error)

func (c *MongoNATSConsumer) Close() error

func (c *MongoNATSConsumer) Start(ctx context.Context, cfg MongoNATSConsumerConfig) error

type MongoNATSConsumerConfig ¶

type MongoNATSConsumerConfig struct {
	URL          string
	StreamName   string
	ConsumerName string
	BatchSize    int
	MaxWait      time.Duration
}

MongoNATSConsumerConfig holds configuration for MongoDB NATS consumer

type ReconciliationBuffer ¶

type ReconciliationBuffer struct {
	// contains filtered or unexported fields
}

ReconciliationBuffer manages buffered events for multiple test runs

func NewReconciliationBuffer(cfg ReconciliationConfig, logger *slog.Logger) *ReconciliationBuffer

func (rb *ReconciliationBuffer) BufferEvent(runID string, event publisher.Event) error

func (rb *ReconciliationBuffer) DeleteBuffer(runID string)

func (rb *ReconciliationBuffer) GetAllBuffers() map[string]*RunBuffer

func (rb *ReconciliationBuffer) GetBuffer(runID string) (*RunBuffer, bool)

func (rb *ReconciliationBuffer) MarkRootSuiteEnd(runID string) error

func (rb *ReconciliationBuffer) SetInactivityCallback(callback func(ctx context.Context, runID string))

func (rb *ReconciliationBuffer) StartCleanupLoop(ctx context.Context)

func (rb *ReconciliationBuffer) StopCleanupLoop()

func (rb *ReconciliationBuffer) UpdateBufferStatus(runID string, status ReconciliationStatus)

type ReconciliationConfig ¶

type ReconciliationConfig struct {
	// Enabled controls whether reconciliation is active
	Enabled bool

	// MaxBufferSize is the maximum number of events to buffer per run
	MaxBufferSize int

	// InactivityTTL is the duration after last event before triggering reconciliation
	InactivityTTL time.Duration

	// MaxPasses is the maximum number of reconciliation passes to attempt
	MaxPasses int

	// PassDelay is the delay between reconciliation passes
	PassDelay time.Duration

	// CleanupInterval is how often to check for inactive buffers
	CleanupInterval time.Duration
}

ReconciliationConfig holds configuration for event reconciliation

func DefaultReconciliationConfig() ReconciliationConfig

func LoadReconciliationConfigFromEnv() ReconciliationConfig

type ReconciliationStatus ¶

type ReconciliationStatus string

ReconciliationStatus represents the state of reconciliation for a run

const (
	StatusPending    ReconciliationStatus = "pending"
	StatusInProgress ReconciliationStatus = "in_progress"
	StatusCompleted  ReconciliationStatus = "completed"
	StatusPartial    ReconciliationStatus = "partial"
	StatusFailed     ReconciliationStatus = "failed"
)

type RunBuffer ¶

type RunBuffer struct {
	RunID        string
	Events       []BufferedEvent
	FirstSeen    time.Time
	LastActivity time.Time // Refreshes on ANY event for this runID

	RootSuiteEndReceived bool
	ReconciliationStatus ReconciliationStatus
	// contains filtered or unexported fields
}

RunBuffer holds buffered events for a specific test run