feat(server): adding event emitter

local-config/tracetest.provision.yaml

@@ -1,2 +1,6 @@
-dataStore:
+---
+type: DataStore
+spec:
+  name: OpenTelemetry Collector
   type: otlp
+  isdefault: true

server/app/app.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"log"
 	"net/http"
-	"time"
 
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
@@ -16,7 +15,6 @@ import (
 	"github.com/kubeshop/tracetest/server/config"
 	"github.com/kubeshop/tracetest/server/config/configresource"
 	"github.com/kubeshop/tracetest/server/config/demoresource"
-	"github.com/kubeshop/tracetest/server/executor"
 	"github.com/kubeshop/tracetest/server/executor/pollingprofile"
 	"github.com/kubeshop/tracetest/server/executor/trigger"
 	httpServer "github.com/kubeshop/tracetest/server/http"
@@ -142,16 +140,16 @@ func (app *App) Start(opts ...appOption) error {
 		testdb.WithDB(db),
 	)
 
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	subscriptionManager := subscription.NewManager()
 	app.subscribeToConfigChanges(subscriptionManager)
 
 	configRepo := configresource.NewRepository(db, configresource.WithPublisher(subscriptionManager))
 	configFromDB := configRepo.Current(ctx)
 
-	if err != nil {
-		log.Fatal(err)
-	}
-
 	tracer, err := tracing.NewTracer(ctx, app.cfg)
 	if err != nil {
 		log.Fatal(err)
@@ -201,7 +199,6 @@ func (app *App) Start(opts ...appOption) error {
 	// worker count. should be configurable
 	rf.tracePoller.Start(5)
 	rf.runner.Start(5)
-	rf.runner.Start(5)
 	rf.transactionRunner.Start(5)
 	rf.assertionRunner.Start(5)
 
@@ -321,68 +318,6 @@ func registerDemosResource(repository *demoresource.Repository, router *mux.Rout
 	provisioner.AddResourceProvisioner(manager)
 }
 
-type facadeConfig interface {
-	PoolingRetryDelay() time.Duration
-	PoolingMaxWaitTimeForTraceDuration() time.Duration
-}
-
-func newRunnerFacades(
-	ppRepo *pollingprofile.Repository,
-	testDB model.Repository,
-	appTracer trace.Tracer,
-	tracer trace.Tracer,
-	subscriptionManager *subscription.Manager,
-	triggerRegistry *trigger.Registry,
-) *runnerFacade {
-
-	execTestUpdater := (executor.CompositeUpdater{}).
-		Add(executor.NewDBUpdater(testDB)).
-		Add(executor.NewSubscriptionUpdater(subscriptionManager))
-
-	assertionRunner := executor.NewAssertionRunner(
-		execTestUpdater,
-		executor.NewAssertionExecutor(tracer),
-		executor.InstrumentedOutputProcessor(tracer),
-		subscriptionManager,
-	)
-
-	pollerExecutor := executor.NewPollerExecutor(
-		ppRepo,
-		tracer,
-		execTestUpdater,
-		tracedb.Factory(testDB),
-		testDB,
-	)
-
-	tracePoller := executor.NewTracePoller(
-		pollerExecutor,
-		ppRepo,
-		execTestUpdater,
-		assertionRunner,
-		subscriptionManager,
-	)
-
-	runner := executor.NewPersistentRunner(
-		triggerRegistry,
-		testDB,
-		execTestUpdater,
-		tracePoller,
-		tracer,
-		subscriptionManager,
-		tracedb.Factory(testDB),
-		testDB,
-	)
-
-	transactionRunner := executor.NewTransactionRunner(runner, testDB, subscriptionManager)
-
-	return &runnerFacade{
-		runner:            runner,
-		transactionRunner: transactionRunner,
-		assertionRunner:   assertionRunner,
-		tracePoller:       tracePoller,
-	}
-}
-
 func getTriggerRegistry(tracer, appTracer trace.Tracer) *trigger.Registry {
 	triggerReg := trigger.NewRegsitry(tracer, appTracer)
 	triggerReg.Add(trigger.HTTP())

server/app/facade.go

@@ -4,7 +4,12 @@ import (
 	"context"
 
 	"github.com/kubeshop/tracetest/server/executor"
+	"github.com/kubeshop/tracetest/server/executor/pollingprofile"
+	"github.com/kubeshop/tracetest/server/executor/trigger"
 	"github.com/kubeshop/tracetest/server/model"
+	"github.com/kubeshop/tracetest/server/subscription"
+	"github.com/kubeshop/tracetest/server/tracedb"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type runnerFacade struct {
@@ -25,3 +30,66 @@ func (rf runnerFacade) RunTransaction(ctx context.Context, tr model.Transaction,
 func (rf runnerFacade) RunAssertions(ctx context.Context, request executor.AssertionRequest) {
 	rf.assertionRunner.RunAssertions(ctx, request)
 }
+
+func newRunnerFacades(
+	ppRepo *pollingprofile.Repository,
+	testDB model.Repository,
+	appTracer trace.Tracer,
+	tracer trace.Tracer,
+	subscriptionManager *subscription.Manager,
+	triggerRegistry *trigger.Registry,
+) *runnerFacade {
+	eventEmitter := executor.NewEventEmitter(testDB, subscriptionManager)
+
+	execTestUpdater := (executor.CompositeUpdater{}).
+		Add(executor.NewDBUpdater(testDB)).
+		Add(executor.NewSubscriptionUpdater(subscriptionManager))
+
+	assertionRunner := executor.NewAssertionRunner(
+		execTestUpdater,
+		executor.NewAssertionExecutor(tracer),
+		executor.InstrumentedOutputProcessor(tracer),
+		subscriptionManager,
+		eventEmitter,
+	)
+
+	pollerExecutor := executor.NewPollerExecutor(
+		ppRepo,
+		tracer,
+		execTestUpdater,
+		tracedb.Factory(testDB),
+		testDB,
+	)
+
+	tracePoller := executor.NewTracePoller(
+		pollerExecutor,
+		ppRepo,
+		execTestUpdater,
+		assertionRunner,
+		subscriptionManager,
+	)
+
+	runner := executor.NewPersistentRunner(
+		triggerRegistry,
+		testDB,
+		execTestUpdater,
+		tracePoller,
+		tracer,
+		subscriptionManager,
+		tracedb.Factory(testDB),
+		testDB,
+	)
+
+	transactionRunner := executor.NewTransactionRunner(
+		runner,
+		testDB,
+		subscriptionManager,
+	)
+
+	return &runnerFacade{
+		runner:            runner,
+		transactionRunner: transactionRunner,
+		assertionRunner:   assertionRunner,
+		tracePoller:       tracePoller,
+	}
+}

server/executor/assertion_runner.go

@@ -46,6 +46,7 @@ func NewAssertionRunner(
 	assertionExecutor AssertionExecutor,
 	op OutputsProcessorFn,
 	subscriptionManager *subscription.Manager,
+	eventEmitter EventEmitter,
 ) AssertionRunner {
 	return &defaultAssertionRunner{
 		outputsProcessor:    op,

server/executor/eventemitter.go

@@ -0,0 +1,38 @@
+package executor
+
+import (
+	"context"
+
+	"github.com/kubeshop/tracetest/server/model"
+)
+
+type EventEmitter interface {
+	Emit(ctx context.Context, event model.TestRunEvent) error
+}
+
+type publisher interface {
+	Publish(eventID string, message any)
+}
+
+type internalEventEmitter struct {
+	repository model.TestRunEventRepository
+	publisher  publisher
+}
+
+func NewEventEmitter(repository model.TestRunEventRepository, publisher publisher) EventEmitter {
+	return &internalEventEmitter{
+		repository: repository,
+		publisher:  publisher,
+	}
+}
+
+func (em *internalEventEmitter) Emit(ctx context.Context, event model.TestRunEvent) error {
+	err := em.repository.CreateTestRunEvent(ctx, event)
+	if err != nil {
+		return err
+	}
+
+	em.publisher.Publish(event.ResourceID(), event)
+
+	return nil
+}