feat: React web UI dashboard

VERSION

@@ -1 +1 @@
-v25.1.1
+v25.2.1

api/server_suite_test.go

@@ -5,10 +5,12 @@ package api
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
+
 	"io"
 	"math/rand"
 	"net/http"
@@ -97,7 +99,7 @@ func getDB() database.Database {
 		_ = os.Setenv("TZ", "") // Use UTC by default :)
 
 		dbHooks := hooks.Init()
-		dbHooks.RegisterHook(datastore.EndpointCreated, func(data interface{}, changelog interface{}) {})
+		dbHooks.RegisterHook(datastore.EndpointCreated, func(ctx context.Context, data interface{}, changelog interface{}) {})
 
 		pDB = db
 	})

cmd/hooks/hooks.go

@@ -232,6 +232,7 @@ func PreRun(app *cli.App, db *postgres.Postgres) func(cmd *cobra.Command, args [
 		if err != nil {
 			return err
 		}
+		// todo(raymond): I don't think the check below needs to exist since we perform the check in the tracer.Init() above.
 		if cfg.Tracer.Type == config.DatadogTracerProvider && !app.Licenser.DatadogTracing() {
 			lo.Error("your instance does not have access to datadog tracing, upgrade to access this feature")
 			_ = app.TracerBackend.Shutdown(context.Background())

cmd/worker/worker.go

@@ -276,6 +276,8 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		a.Licenser,
 		featureFlag,
 		net.LoggerOption(lo),
+		net.TracerOption(a.TracerBackend),
+		net.DetailedTraceOption(true),
 		net.ProxyOption(cfg.Server.HTTP.HttpProxy),
 		net.AllowListOption(cfg.Dispatcher.AllowList),
 		net.BlockListOption(cfg.Dispatcher.BlockList),
@@ -364,8 +366,8 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		attemptRepo,
 		circuitBreakerManager,
 		featureFlag,
-		a.TracerBackend,
-	), newTelemetry)
+		a.TracerBackend),
+		newTelemetry)
 
 	consumer.RegisterHandlers(convoy.CreateEventProcessor, task.ProcessEventCreation(
 		defaultCh,
@@ -375,7 +377,10 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		eventDeliveryRepo,
 		a.Queue,
 		subRepo,
-		deviceRepo, a.Licenser), newTelemetry)
+		deviceRepo,
+		a.Licenser,
+		a.TracerBackend),
+		newTelemetry)
 
 	consumer.RegisterHandlers(convoy.RetryEventProcessor, task.ProcessRetryEventDelivery(
 		endpointRepo,
@@ -388,8 +393,8 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		attemptRepo,
 		circuitBreakerManager,
 		featureFlag,
-		a.TracerBackend,
-	), newTelemetry)
+		a.TracerBackend),
+		newTelemetry)
 
 	consumer.RegisterHandlers(convoy.CreateBroadcastEventProcessor, task.ProcessBroadcastEventCreation(
 		broadcastCh,
@@ -400,7 +405,9 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		a.Queue,
 		subRepo,
 		deviceRepo,
-		a.Licenser), newTelemetry)
+		a.Licenser,
+		a.TracerBackend),
+		newTelemetry)
 
 	consumer.RegisterHandlers(convoy.CreateDynamicEventProcessor, task.ProcessDynamicEventCreation(
 		dynamicCh,
@@ -410,7 +417,10 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		eventDeliveryRepo,
 		a.Queue,
 		subRepo,
-		deviceRepo, a.Licenser), newTelemetry)
+		deviceRepo,
+		a.Licenser,
+		a.TracerBackend),
+		newTelemetry)
 
 	if a.Licenser.RetentionPolicy() {
 		consumer.RegisterHandlers(convoy.RetentionPolicies, task.RetentionPolicies(rd, ret), nil)
@@ -425,7 +435,10 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		eventDeliveryRepo,
 		a.Queue,
 		subRepo,
-		deviceRepo, a.Licenser), newTelemetry)
+		deviceRepo,
+		a.Licenser,
+		a.TracerBackend),
+		newTelemetry)
 
 	consumer.RegisterHandlers(convoy.MonitorTwitterSources, task.MonitorTwitterSources(a.DB, a.Queue, rd), nil)
 

database/hooks/hooks.go

@@ -1,13 +1,14 @@
 package hooks
 
 import (
+	"context"
 	"errors"
 	"sync/atomic"
 
 	"github.com/frain-dev/convoy/datastore"
 )
 
-type hookMap map[datastore.HookEventType]func(data interface{}, changelog interface{})
+type hookMap map[datastore.HookEventType]func(ctx context.Context, data interface{}, changelog interface{})
 
 type Hook struct {
 	fns hookMap
@@ -30,13 +31,13 @@ func Init() *Hook {
 	return &Hook{fns: hookMap{}}
 }
 
-func (h *Hook) Fire(eventType datastore.HookEventType, data interface{}, changelog interface{}) {
+func (h *Hook) Fire(ctx context.Context, eventType datastore.HookEventType, data interface{}, changelog interface{}) {
 	if fn, ok := h.fns[eventType]; ok {
-		fn(data, changelog)
+		fn(ctx, data, changelog)
 	}
 }
 
-func (h *Hook) RegisterHook(eventType datastore.HookEventType, fn func(data interface{}, changelog interface{})) {
+func (h *Hook) RegisterHook(eventType datastore.HookEventType, fn func(ctx context.Context, data interface{}, changelog interface{})) {
 	h.fns[eventType] = fn
 	hookSingleton.Store(h)
 }

database/listener/endpoint_listener.go

@@ -1,6 +1,7 @@
 package listener
 
 import (
+	"context"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/pkg/log"
 	"github.com/frain-dev/convoy/queue"
@@ -16,26 +17,26 @@ func NewEndpointListener(queue queue.Queuer, projectRepo datastore.ProjectReposi
 	return &EndpointListener{mEvent: mEvent}
 }
 
-func (e *EndpointListener) AfterCreate(data interface{}, _ interface{}) {
-	e.metaEvent(string(datastore.EndpointCreated), data)
+func (e *EndpointListener) AfterCreate(ctx context.Context, data interface{}, _ interface{}) {
+	e.metaEvent(ctx, datastore.EndpointCreated, data)
 }
 
-func (e *EndpointListener) AfterUpdate(data interface{}, _ interface{}) {
-	e.metaEvent(string(datastore.EndpointUpdated), data)
+func (e *EndpointListener) AfterUpdate(ctx context.Context, data interface{}, _ interface{}) {
+	e.metaEvent(ctx, datastore.EndpointUpdated, data)
 }
 
-func (e *EndpointListener) AfterDelete(data interface{}, _ interface{}) {
-	e.metaEvent(string(datastore.EndpointDeleted), data)
+func (e *EndpointListener) AfterDelete(ctx context.Context, data interface{}, _ interface{}) {
+	e.metaEvent(ctx, datastore.EndpointDeleted, data)
 }
 
-func (e *EndpointListener) metaEvent(eventType string, data interface{}) {
+func (e *EndpointListener) metaEvent(ctx context.Context, eventType datastore.HookEventType, data interface{}) {
 	endpoint, ok := data.(*datastore.Endpoint)
 	if !ok {
 		log.Errorf("invalid type for event - %s", eventType)
 		return
 	}
 
-	if err := e.mEvent.Run(eventType, endpoint.ProjectID, endpoint); err != nil {
+	if err := e.mEvent.Run(ctx, string(eventType), endpoint.ProjectID, endpoint); err != nil {
 		log.WithError(err).Error("endpoint meta event failed")
 	}
 }

database/listener/event_delivery_listener.go

@@ -47,15 +47,15 @@ func NewEventDeliveryListener(queue queue.Queuer, projectRepo datastore.ProjectR
 	return &EventDeliveryListener{mEvent: mEvent, attemptsRepo: attemptsRepo}
 }
 
-func (e *EventDeliveryListener) AfterUpdate(data interface{}, _ interface{}) {
+func (e *EventDeliveryListener) AfterUpdate(ctx context.Context, data interface{}, _ interface{}) {
 	eventDelivery, ok := data.(*datastore.EventDelivery)
 	if !ok {
 		log.Error("invalid type for event - eventdelivery.updated")
 		return
 	}
 
 	mEventDelivery := getMetaEventDelivery(eventDelivery)
-	attempts, err := e.attemptsRepo.FindDeliveryAttempts(context.Background(), mEventDelivery.UID)
+	attempts, err := e.attemptsRepo.FindDeliveryAttempts(ctx, mEventDelivery.UID)
 	if err != nil {
 		log.WithError(err).Error("event delivery meta event failed")
 	}
@@ -65,14 +65,14 @@ func (e *EventDeliveryListener) AfterUpdate(data interface{}, _ interface{}) {
 	}
 
 	if eventDelivery.Status == datastore.SuccessEventStatus {
-		err := e.mEvent.Run(string(datastore.EventDeliverySuccess), eventDelivery.ProjectID, mEventDelivery)
+		err = e.mEvent.Run(ctx, string(datastore.EventDeliverySuccess), eventDelivery.ProjectID, mEventDelivery)
 		if err != nil {
 			log.WithError(err).Error("event delivery meta event failed")
 		}
 	}
 
 	if eventDelivery.Status == datastore.FailureEventStatus {
-		err := e.mEvent.Run(string(datastore.EventDeliveryFailed), eventDelivery.ProjectID, mEventDelivery)
+		err = e.mEvent.Run(ctx, string(datastore.EventDeliveryFailed), eventDelivery.ProjectID, mEventDelivery)
 		if err != nil {
 			log.WithError(err).Error("event delivery meta event failed")
 		}

database/listener/project_listener.go

@@ -1,6 +1,7 @@
 package listener
 
 import (
+	"context"
 	"encoding/json"
 	"time"
 
@@ -19,11 +20,11 @@ func NewProjectListener(queue queue.Queuer) *ProjectListener {
 	return &ProjectListener{queue: queue}
 }
 
-func (e *ProjectListener) AfterUpdate(data interface{}, changelog interface{}) {
-	e.run(string(datastore.ProjectUpdated), data, changelog)
+func (e *ProjectListener) AfterUpdate(ctx context.Context, data interface{}, changelog interface{}) {
+	e.run(ctx, datastore.ProjectUpdated, data, changelog)
 }
 
-func (e *ProjectListener) run(eventType string, data interface{}, changelog interface{}) {
+func (e *ProjectListener) run(_ context.Context, eventType datastore.HookEventType, data interface{}, changelog interface{}) {
 	project, ok := data.(*datastore.Project)
 	if !ok {
 		log.Errorf("invalid type for project - %s", eventType)

database/postgres/endpoint.go

@@ -305,7 +305,7 @@ func (e *endpointRepo) CreateEndpoint(ctx context.Context, endpoint *datastore.E
 		return ErrEndpointNotCreated
 	}
 
-	go e.hook.Fire(datastore.EndpointCreated, endpoint, nil)
+	go e.hook.Fire(context.Background(), datastore.EndpointCreated, endpoint, nil)
 
 	return nil
 }
@@ -418,7 +418,7 @@ func (e *endpointRepo) UpdateEndpoint(ctx context.Context, endpoint *datastore.E
 		return ErrEndpointNotUpdated
 	}
 
-	go e.hook.Fire(datastore.EndpointUpdated, endpoint, nil)
+	go e.hook.Fire(context.Background(), datastore.EndpointUpdated, endpoint, nil)
 	return nil
 }
 
@@ -467,7 +467,7 @@ func (e *endpointRepo) DeleteEndpoint(ctx context.Context, endpoint *datastore.E
 		return err
 	}
 
-	go e.hook.Fire(datastore.EndpointDeleted, endpoint, nil)
+	go e.hook.Fire(context.Background(), datastore.EndpointDeleted, endpoint, nil)
 	return nil
 }
 

database/postgres/event_delivery.go

@@ -555,7 +555,8 @@ func (e *eventDeliveryRepo) UpdateEventDeliveryMetadata(ctx context.Context, pro
 		return ErrEventDeliveryAttemptsNotUpdated
 	}
 
-	go e.hook.Fire(datastore.EventDeliveryUpdated, delivery, nil)
+	e.hook.Fire(ctx, datastore.EventDeliveryUpdated, delivery, nil)
+
 	return nil
 }
 

database/postgres/postgres_test.go

@@ -71,7 +71,7 @@ func getDB(t *testing.T) (database.Database, func()) {
 		var err error
 
 		dbHooks := hooks.Init()
-		dbHooks.RegisterHook(datastore.EndpointCreated, func(data interface{}, changelog interface{}) {})
+		dbHooks.RegisterHook(datastore.EndpointCreated, func(ctx context.Context, data interface{}, changelog interface{}) {})
 
 		_db, err = NewDB(getConfig())
 		require.NoError(t, err)

database/postgres/project.go

@@ -423,7 +423,8 @@ func (p *projectRepo) UpdateProject(ctx context.Context, project *datastore.Proj
 		return err
 	}
 
-	go p.hook.Fire(datastore.ProjectUpdated, project, changelog)
+	go p.hook.Fire(context.Background(), datastore.ProjectUpdated, project, changelog)
+
 	return nil
 }
 

datastore/models.go

@@ -6,12 +6,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	cb "github.com/frain-dev/convoy/pkg/circuit_breaker"
 	"math"
 	"net/http"
 	"strings"
 	"time"
 
+	cb "github.com/frain-dev/convoy/pkg/circuit_breaker"
+
 	"github.com/frain-dev/convoy/pkg/flatten"
 
 	"github.com/oklog/ulid/v2"
@@ -963,6 +964,7 @@ type EventDelivery struct {
 	Headers        httpheader.HTTPHeader `json:"headers" db:"headers"`
 	URLQueryParams string                `json:"url_query_params" db:"url_query_params"`
 	IdempotencyKey string                `json:"idempotency_key" db:"idempotency_key"`
+
 	// Deprecated: Latency is deprecated.
 	Latency        string    `json:"latency" db:"latency"`
 	LatencySeconds float64   `json:"latency_seconds" db:"latency_seconds"`

ee/VERSION

@@ -1 +1 @@
-v25.1.1 Enterprise Edition
+v25.2.1 Enterprise Edition

generate.go

@@ -11,3 +11,4 @@ package convoy
 //go:generate mockgen --source internal/pkg/dedup/dedup.go --destination mocks/dedup.go -package mocks
 //go:generate mockgen --source internal/pkg/memorystore/table.go --destination mocks/table.go -package mocks
 //go:generate mockgen --source internal/pkg/license/license.go --destination mocks/license.go -package mocks
+//go:generate mockgen --source internal/pkg/tracer/tracer.go --destination mocks/tracer.go -package mocks

internal/pkg/keys/hcpvault_test.go

@@ -37,7 +37,6 @@ func TestNewHCPVaultKeyManagerEnv(t *testing.T) {
 	key, err := h.GetCurrentKeyFromCache()
 	assert.Nil(t, err)
 	assert.NotEmpty(t, key)
-	t.Logf("Retrieved key: %s", key)
 
 	// Happy path for setting a key
 	newKey := "from-test-" + time.Now().String()

internal/pkg/socket/client_test.go

@@ -16,10 +16,9 @@ import (
 
 func provideClient(r *Repo, c WebSocketConnection) *Client {
 	return &Client{
-		conn:       c,
-		deviceID:   "1234",
-		deviceRepo: r.DeviceRepo,
-		// EventTypes:        []string{"*"},
+		conn:              c,
+		deviceID:          "1234",
+		deviceRepo:        r.DeviceRepo,
 		eventDeliveryRepo: r.EventDeliveryRepo,
 		Device: &datastore.Device{
 			UID:        "1234",
@@ -349,7 +348,6 @@ func TestParseTime(t *testing.T) {
 			client := provideClient(r, c)
 
 			since, err := client.parseTime(tt.args.message)
-			t.Log(since)
 
 			if tt.args.wantErr {
 				require.Error(t, tt.args.err, err)