Repository: frain-dev/convoy

Issue:
[Bug report]: Add jobID to payload passed to workers


Hints from issue comments:
@linear: <p><a href="https://linear.app/convoy-engineering/issue/PDE-574/bug-report-add-jobid-to-payload-passed-to-workers">PDE-574</a></p>

Solution patch:
diff --git a/.mise-tasks/dashboard/build.sh b/.mise-tasks/dashboard/build.sh
index 3b191ef4..321eb2cc 100755
--- a/.mise-tasks/dashboard/build.sh
+++ b/.mise-tasks/dashboard/build.sh
@@ -2,6 +2,7 @@
 
 #MISE description="Build the Angular dashboard application"
 #MISE dir="{{ config_root }}"
+#MISE sources=["web/ui/dashboard/**/*"]
 
 set -e
 
diff --git a/.mise-tasks/lint/server.sh b/.mise-tasks/lint/server.sh
index c868920e..63c41686 100755
--- a/.mise-tasks/lint/server.sh
+++ b/.mise-tasks/lint/server.sh
@@ -7,4 +7,5 @@ set -e
 
 echo "Linting server"
 
+go vet ./...
 exec gofmt -s -l .
diff --git a/.mise-tasks/start/agent b/.mise-tasks/start/agent
index 16245ee0..14fcb426 100755
--- a/.mise-tasks/start/agent
+++ b/.mise-tasks/start/agent
@@ -7,4 +7,4 @@
 set -e
 
 echo "🚀 Starting Agent"
-exec ./bin/convoy server
+exec ./bin/convoy agent
diff --git a/Makefile b/Makefile
index 3e7955d5..57cdcad0 100644
--- a/Makefile
+++ b/Makefile
@@ -20,7 +20,7 @@ test:
 
 # E2E Tests - Run individually for reliability and reproducibility
 test_e2e:
-	@echo "Running E2E tests individually for maximum reliability..."
+	@echo "Running E2E tests individually..."
 	@go test -v ./e2e/... -run TestE2E_DirectEvent_AllSubscriptions -timeout 2m
 	@go test -v ./e2e/... -run TestE2E_DirectEvent_MustMatchSubscription -timeout 2m
 	@go test -v ./e2e/... -run TestE2E_FanOutEvent_AllSubscriptions -timeout 2m
@@ -29,6 +29,17 @@ test_e2e:
 	@go test -v ./e2e/... -run TestE2E_FormEndpoint_WithCustomHeaders -timeout 2m
 	@go test -v ./e2e/... -run TestE2E_OAuth2_SharedSecret -timeout 2m
 	@go test -v ./e2e/... -run TestE2E_OAuth2_ClientAssertion -timeout 2m
+	@echo "Running Job ID E2E tests..."
+	@go test -v ./e2e/... -run TestE2E_SingleEvent_JobID_Format -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_SingleEvent_JobID_Deduplication -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_FanoutEvent_JobID_Format -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_FanoutEvent_MultipleOwners -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_BroadcastEvent_JobID_Format -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_BroadcastEvent_AllSubscribers -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_DynamicEvent_JobID_Format -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_DynamicEvent_MultipleEventTypes -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_ReplayEvent_JobID_Format -timeout 2m
+	@go test -v ./e2e/... -run TestE2E_ReplayEvent_MultipleReplays -timeout 2m
 	@echo "✅ All E2E tests passed!"
 
 # Run all E2E tests together (may be flaky, use test_e2e for CI)
diff --git a/api/handlers/endpoint.go b/api/handlers/endpoint.go
index c8c8b60d..4d6ac84a 100644
--- a/api/handlers/endpoint.go
+++ b/api/handlers/endpoint.go
@@ -607,7 +607,7 @@ func (h *Handler) ActivateEndpoint(w http.ResponseWriter, r *http.Request) {
 //	@Id				TestOAuth2Connection
 //	@Accept			json
 //	@Produce		json
-//	@Param			projectID	path		string					true	"Project ID"
+//	@Param			projectID	path		string						true	"Project ID"
 //	@Param			oauth2		body		models.TestOAuth2Request	true	"OAuth2 Configuration"
 //	@Success		200			{object}	util.ServerResponse{data=models.TestOAuth2Response}
 //	@Failure		400,401,404	{object}	util.ServerResponse{data=Stub}
diff --git a/api/handlers/event.go b/api/handlers/event.go
index 331a20df..c99e2166 100644
--- a/api/handlers/event.go
+++ b/api/handlers/event.go
@@ -68,9 +68,12 @@ func (h *Handler) CreateEndpointEvent(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	id := ulid.Make().String()
+	jobId := queue.JobId{ProjectID: projectID, ResourceID: id}.SingleJobId()
 	e := task.CreateEvent{
+		JobID: jobId,
 		Params: task.CreateEventTaskParams{
-			UID:            ulid.Make().String(),
+			UID:            id,
 			ProjectID:      projectID,
 			EndpointID:     newMessage.EndpointID,
 			EventType:      newMessage.EventType,
@@ -88,11 +91,9 @@ func (h *Handler) CreateEndpointEvent(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	jobId := fmt.Sprintf("single:%s:%s", e.Params.ProjectID, e.Params.UID)
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
-		Delay:   0,
 	}
 
 	err = h.A.Queue.Write(convoy.CreateEventProcessor, convoy.CreateEventQueue, job)
diff --git a/api/ingest.go b/api/ingest.go
index b6753af3..b044375a 100644
--- a/api/ingest.go
+++ b/api/ingest.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -223,7 +222,9 @@ func (a *ApplicationHandler) IngestEvent(w http.ResponseWriter, r *http.Request)
 
 	event.Headers["X-Convoy-Source-Id"] = []string{source.MaskID}
 
+	jobId := queue.JobId{ProjectID: event.ProjectID, ResourceID: event.UID}.SingleJobId()
 	createEvent := task.CreateEvent{
+		JobID: jobId,
 		Event: event,
 	}
 
@@ -234,7 +235,6 @@ func (a *ApplicationHandler) IngestEvent(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	jobId := fmt.Sprintf("single:%s:%s", event.ProjectID, event.UID)
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
diff --git a/cmd/worker/notifications.go b/cmd/worker/notifications.go
index 76a96edb..7502fe39 100644
--- a/cmd/worker/notifications.go
+++ b/cmd/worker/notifications.go
@@ -3,6 +3,8 @@ package worker
 import (
 	"fmt"
 
+	"github.com/oklog/ulid/v2"
+
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/email"
@@ -82,6 +84,9 @@ func enqueueEmail(q queue.Queuer, emailMsg *email.Message) error {
 	if err != nil {
 		return err
 	}
-	job := &queue.Job{Payload: bytes, Delay: 0}
+	job := &queue.Job{
+		ID:      ulid.Make().String(),
+		Payload: bytes,
+	}
 	return q.Write(convoy.NotificationProcessor, convoy.DefaultQueue, job)
 }
diff --git a/cmd/worker/worker.go b/cmd/worker/worker.go
index 78fd1365..3325037a 100644
--- a/cmd/worker/worker.go
+++ b/cmd/worker/worker.go
@@ -133,6 +133,15 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration) erro
 
 	// register worker.
 	consumer := worker.NewConsumer(ctx, cfg.ConsumerPoolSize, q, lo, lvl)
+
+	// Inject job tracker if set (for E2E tests)
+	if a.JobTracker != nil {
+		if tracker, ok := a.JobTracker.(worker.JobTracker); ok {
+			consumer.SetJobTracker(tracker)
+			lo.Info("Job tracker injected into worker consumer")
+		}
+	}
+
 	projectRepo := postgres.NewProjectRepo(a.DB)
 	metaEventRepo := postgres.NewMetaEventRepo(a.DB)
 	endpointRepo := postgres.NewEndpointRepo(a.DB)
diff --git a/docs/docs.go b/docs/docs.go
index 74838ff1..db3bbc8e 100644
--- a/docs/docs.go
+++ b/docs/docs.go
@@ -1,4 +1,4 @@
-// Package docs Code generated by swaggo/swag at 2025-12-01 21:46:56.904172 +0100 CET m=+1.675796959. DO NOT EDIT
+// Package docs Code generated by swaggo/swag at 2025-12-02 16:33:07.564697 +0100 CET m=+1.662062084. DO NOT EDIT
 package docs
 
 import "github.com/swaggo/swag"
diff --git a/e2e/helpers.go b/e2e/helpers.go
index a114ed49..f08f18f0 100644
--- a/e2e/helpers.go
+++ b/e2e/helpers.go
@@ -233,6 +233,24 @@ func SendEventViaSDK(t *testing.T, c *convoy.Client, endpointUID, eventType, tra
 	require.NoError(t, err)
 }
 
+// SendEventWithIdempotencyKey sends an event with a specific idempotency key using the Convoy Go SDK
+func SendEventWithIdempotencyKey(t *testing.T, c *convoy.Client, endpointUID, eventType, traceID, idempotencyKey string) {
+	t.Helper()
+
+	event := fmt.Sprintf(`{"traceId": %q}`, traceID)
+	payload := []byte(event)
+
+	body := &convoy.CreateEventRequest{
+		EventType:      eventType,
+		EndpointID:     endpointUID,
+		IdempotencyKey: idempotencyKey,
+		Data:           payload,
+	}
+
+	err := c.Events.Create(t.Context(), body)
+	require.NoError(t, err)
+}
+
 // SendFanoutEventViaSDK sends a fanout event using the Convoy Go SDK
 func SendFanoutEventViaSDK(t *testing.T, c *convoy.Client, ownerID, eventType, traceID string) {
 	t.Helper()
diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go
index 9bd59948..2be98bc0 100644
--- a/internal/notifications/notifications.go
+++ b/internal/notifications/notifications.go
@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/oklog/ulid/v2"
+
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/email"
@@ -104,7 +106,10 @@ func SendEndpointNotification(
 			continue
 		}
 
-		job := &queue.Job{Payload: buf}
+		job := &queue.Job{
+			ID:      ulid.Make().String(),
+			Payload: buf,
+		}
 
 		err = q.Write(convoy.NotificationProcessor, convoy.DefaultQueue, job)
 		if err != nil {
diff --git a/internal/pkg/cli/cli.go b/internal/pkg/cli/cli.go
index 249f4cd5..7e226c58 100644
--- a/internal/pkg/cli/cli.go
+++ b/internal/pkg/cli/cli.go
@@ -27,6 +27,10 @@ type App struct {
 	Licenser license.Licenser
 
 	TracerBackend tracer.Backend
+
+	// JobTracker is an optional field used only in E2E tests
+	// to capture job IDs for verification
+	JobTracker interface{}
 }
 
 type ConvoyCli struct {
diff --git a/internal/pkg/pubsub/ingest.go b/internal/pkg/pubsub/ingest.go
index ff64dcba..fbfeeabf 100644
--- a/internal/pkg/pubsub/ingest.go
+++ b/internal/pkg/pubsub/ingest.go
@@ -235,9 +235,12 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 	messageType := headers[common.ConvoyMessageTypeHeader]
 	switch messageType {
 	case "single":
+		id := ulid.Make().String()
+		jobId := queue.JobId{ProjectID: source.ProjectID, ResourceID: id}.SingleJobId()
 		ce := task.CreateEvent{
+			JobID: jobId,
 			Params: task.CreateEventTaskParams{
-				UID:            ulid.Make().String(),
+				UID:            id,
 				SourceID:       source.UID,
 				ProjectID:      source.ProjectID,
 				EndpointID:     convoyEvent.EndpointID,
@@ -269,7 +272,6 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 			return err
 		}
 
-		jobId := fmt.Sprintf("single:%s:%s", ce.Params.ProjectID, ce.Params.UID)
 		job := &queue.Job{
 			ID:      jobId,
 			Payload: eventByte,
@@ -281,9 +283,12 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 			return err
 		}
 	case "fanout":
+		id := ulid.Make().String()
+		jobId := queue.JobId{ProjectID: source.ProjectID, ResourceID: id}.FanOutJobId()
 		ce := task.CreateEvent{
+			JobID: jobId,
 			Params: task.CreateEventTaskParams{
-				UID:            ulid.Make().String(),
+				UID:            id,
 				CustomHeaders:  headers,
 				SourceID:       source.UID,
 				Data:           convoyEvent.Data,
@@ -307,7 +312,6 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 			return err
 		}
 
-		jobId := fmt.Sprintf("fanout:%s:%s", source.ProjectID, ce.Params.UID)
 		job := &queue.Job{
 			ID:      jobId,
 			Payload: eventByte,
@@ -320,8 +324,9 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 		}
 	case "broadcast":
 		eventId := ulid.Make().String()
-		jobId := fmt.Sprintf("broadcast:%s:%s", source.ProjectID, eventId)
+		jobId := queue.JobId{ProjectID: source.ProjectID, ResourceID: eventId}.BroadcastJobId()
 		broadcastEvent := models.BroadcastEvent{
+			JobID:          jobId,
 			EventID:        eventId,
 			ProjectID:      source.ProjectID,
 			SourceID:       source.UID,
@@ -348,7 +353,7 @@ func (i *Ingest) handler(ctx context.Context, source *datastore.Source, msg stri
 			return err
 		}
 	default:
-		err := fmt.Errorf("%s isn't a valid pubsub message type, it should be one of single, fanout or broadcast", messageType)
+		err = fmt.Errorf("%s isn't a valid pubsub message type, it should be one of single, fanout or broadcast", messageType)
 		log.Error(err)
 		return err
 	}
diff --git a/queue/queue.go b/queue/queue.go
index c1764f92..08234a0a 100644
--- a/queue/queue.go
+++ b/queue/queue.go
@@ -1,6 +1,7 @@
 package queue
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/frain-dev/convoy"
@@ -26,3 +27,36 @@ type QueueOptions struct {
 	RedisAddress      []string
 	PrometheusAddress string
 }
+
+type JobId struct {
+	ProjectID  string
+	ResourceID string
+}
+
+func (j JobId) SingleJobId() string {
+	return fmt.Sprintf("single:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) MetaJobId() string {
+	return fmt.Sprintf("meta:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) DynamicJobId() string {
+	return fmt.Sprintf("dynamic:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) BroadcastJobId() string {
+	return fmt.Sprintf("broadcast:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) FanOutJobId() string {
+	return fmt.Sprintf("fanout:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) ReplayJobId() string {
+	return fmt.Sprintf("replay:%s:%s", j.ProjectID, j.ResourceID)
+}
+
+func (j JobId) MatchSubsJobId() string {
+	return fmt.Sprintf("match_subs:%s:%s", j.ProjectID, j.ResourceID)
+}
diff --git a/services/create_broadcast_event.go b/services/create_broadcast_event.go
index ba99b892..11d8a59f 100644
--- a/services/create_broadcast_event.go
+++ b/services/create_broadcast_event.go
@@ -2,7 +2,6 @@ package services
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"time"
 
@@ -28,9 +27,12 @@ func (e *CreateBroadcastEventService) Run(ctx context.Context) error {
 		return &ServiceError{ErrMsg: "an error occurred while creating broadcast event - invalid project"}
 	}
 
-	e.BroadcastEvent.EventID = ulid.Make().String()
+	id := ulid.Make().String()
+	jobId := queue.JobId{ProjectID: e.Project.UID, ResourceID: id}.BroadcastJobId()
+
+	e.BroadcastEvent.EventID = id
+	e.BroadcastEvent.JobID = jobId
 	e.BroadcastEvent.ProjectID = e.Project.UID
-	jobId := fmt.Sprintf("broadcast:%s:%s", e.BroadcastEvent.ProjectID, e.BroadcastEvent.EventID)
 	e.BroadcastEvent.AcknowledgedAt = time.Now()
 
 	taskName := convoy.CreateBroadcastEventProcessor
@@ -43,7 +45,6 @@ func (e *CreateBroadcastEventService) Run(ctx context.Context) error {
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
-		Delay:   0,
 	}
 
 	err = e.Queue.Write(taskName, convoy.CreateEventQueue, job)
diff --git a/services/create_dynamic_event.go b/services/create_dynamic_event.go
index 67c4f7c1..b12c39df 100644
--- a/services/create_dynamic_event.go
+++ b/services/create_dynamic_event.go
@@ -2,11 +2,10 @@ package services
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"time"
 
-	"github.com/google/uuid"
+	"github.com/oklog/ulid/v2"
 
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/api/models"
@@ -28,10 +27,12 @@ func (e *CreateDynamicEventService) Run(ctx context.Context) error {
 	if e.Project == nil {
 		return &ServiceError{ErrMsg: "an error occurred while creating dynamic event - invalid project"}
 	}
+	id := ulid.Make().String()
+	jobId := queue.JobId{ProjectID: e.Project.UID, ResourceID: id}.DynamicJobId()
 
-	e.DynamicEvent.EventID = uuid.NewString()
+	e.DynamicEvent.EventID = id
+	e.DynamicEvent.JobID = jobId
 	e.DynamicEvent.ProjectID = e.Project.UID
-	jobId := fmt.Sprintf("dynamic:%s:%s", e.DynamicEvent.ProjectID, e.DynamicEvent.EventID)
 	e.DynamicEvent.AcknowledgedAt = time.Now()
 
 	if len(e.DynamicEvent.EventTypes) == 0 {
@@ -48,7 +49,6 @@ func (e *CreateDynamicEventService) Run(ctx context.Context) error {
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
-		Delay:   0,
 	}
 
 	err = e.Queue.Write(taskName, convoy.CreateEventQueue, job)
diff --git a/services/create_fanout_event.go b/services/create_fanout_event.go
index 7fc5cc8a..40da4c51 100644
--- a/services/create_fanout_event.go
+++ b/services/create_fanout_event.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"time"
 
 	"github.com/oklog/ulid/v2"
@@ -102,13 +101,14 @@ func (e *CreateFanoutEventService) Run(ctx context.Context) (event *datastore.Ev
 	return event, nil
 }
 
-func createEvent(ctx context.Context, endpoints []datastore.Endpoint, newMessage *newEvent, g *datastore.Project, queuer queue.Queuer) (*datastore.Event, error) {
+func createEvent(ctx context.Context, endpoints []datastore.Endpoint, newMessage *newEvent, project *datastore.Project, queuer queue.Queuer) (*datastore.Event, error) {
 	endpointIDs := make([]string, 0, len(endpoints))
 
 	for _, endpoint := range endpoints {
 		endpointIDs = append(endpointIDs, endpoint.UID)
 	}
 
+	jobId := queue.JobId{ProjectID: project.UID, ResourceID: newMessage.UID}.FanOutJobId()
 	event := &datastore.Event{
 		UID:              newMessage.UID,
 		EventType:        datastore.EventType(newMessage.EventType),
@@ -118,16 +118,17 @@ func createEvent(ctx context.Context, endpoints []datastore.Endpoint, newMessage
 		IsDuplicateEvent: newMessage.IsDuplicate,
 		Headers:          getCustomHeaders(newMessage.CustomHeaders),
 		Endpoints:        endpointIDs,
-		ProjectID:        g.UID,
+		ProjectID:        project.UID,
 		AcknowledgedAt:   null.TimeFrom(time.Now()),
 	}
 
-	if (g.Config == nil || g.Config.Strategy == nil) ||
-		(g.Config.Strategy != nil && g.Config.Strategy.Type != datastore.LinearStrategyProvider && g.Config.Strategy.Type != datastore.ExponentialStrategyProvider) {
+	if (project.Config == nil || project.Config.Strategy == nil) ||
+		(project.Config.Strategy != nil && project.Config.Strategy.Type != datastore.LinearStrategyProvider && project.Config.Strategy.Type != datastore.ExponentialStrategyProvider) {
 		return nil, &ServiceError{ErrMsg: "retry strategy not defined in configuration"}
 	}
 
 	e := task.CreateEvent{
+		JobID:              jobId,
 		Event:              event,
 		CreateSubscription: !util.IsStringEmpty(newMessage.EndpointID),
 	}
@@ -137,11 +138,9 @@ func createEvent(ctx context.Context, endpoints []datastore.Endpoint, newMessage
 		return nil, &ServiceError{ErrMsg: err.Error()}
 	}
 
-	jobId := fmt.Sprintf("fanout:%s:%s", event.ProjectID, event.UID)
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
-		Delay:   0,
 	}
 	err = queuer.Write(convoy.CreateEventProcessor, convoy.CreateEventQueue, job)
 	if err != nil {
diff --git a/services/create_meta_event.go b/services/create_meta_event.go
index ea9b9ca7..6c3c387c 100644
--- a/services/create_meta_event.go
+++ b/services/create_meta_event.go
@@ -3,7 +3,6 @@ package services
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"time"
 
 	"github.com/oklog/ulid/v2"
@@ -96,7 +95,7 @@ func (m *MetaEvent) Run(ctx context.Context, eventType, projectID string, data i
 		return err
 	}
 
-	jobId := fmt.Sprintf("meta:%s:%s", metaEvent.ProjectID, metaEvent.UID)
+	jobId := queue.JobId{ProjectID: metaEvent.ProjectID, ResourceID: metaEvent.UID}.MetaJobId()
 	err = m.queue.Write(convoy.MetaEventProcessor, convoy.MetaEventQueue, &queue.Job{
 		ID:      jobId,
 		Payload: bytes,
diff --git a/services/invite_user.go b/services/invite_user.go
index b5431258..e39230b9 100644
--- a/services/invite_user.go
+++ b/services/invite_user.go
@@ -111,8 +111,8 @@ func sendInviteEmail(iv *datastore.OrganisationInvite, user *datastore.User, org
 	}
 
 	job := &queue.Job{
+		ID:      iv.UID,
 		Payload: bytes,
-		Delay:   0,
 	}
 
 	err = queuer.Write(convoy.EmailProcessor, convoy.DefaultQueue, job)
diff --git a/services/meta_event_service.go b/services/meta_event_service.go
index 2ac1145c..4e63bafb 100644
--- a/services/meta_event_service.go
+++ b/services/meta_event_service.go
@@ -36,7 +36,7 @@ func (m *MetaEventService) Run(ctx context.Context, metaEvent *datastore.MetaEve
 		return err
 	}
 
-	jobId := fmt.Sprintf("meta:%s:%s", metaEvent.ProjectID, metaEvent.UID)
+	jobId := queue.JobId{ProjectID: metaEvent.ProjectID, ResourceID: metaEvent.UID}.MetaJobId()
 	err = m.Queue.Write(convoy.MetaEventProcessor, convoy.MetaEventQueue, &queue.Job{
 		ID:      jobId,
 		Payload: bytes,
diff --git a/services/register_user.go b/services/register_user.go
index 566317f6..cf60a17f 100644
--- a/services/register_user.go
+++ b/services/register_user.go
@@ -137,8 +137,8 @@ func queueEmail(ctx context.Context, em *email.Message, q queue.Queuer) error {
 	}
 
 	job := &queue.Job{
+		ID:      ulid.Make().String(),
 		Payload: bytes,
-		Delay:   0,
 	}
 
 	err = q.Write(convoy.EmailProcessor, convoy.DefaultQueue, job)
diff --git a/services/replay_event.go b/services/replay_event.go
index 5ab273bd..fcb79987 100644
--- a/services/replay_event.go
+++ b/services/replay_event.go
@@ -2,7 +2,6 @@ package services
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"gopkg.in/guregu/null.v4"
@@ -24,12 +23,14 @@ type ReplayEventService struct {
 }
 
 func (e *ReplayEventService) Run(ctx context.Context) error {
-	createEvent := task.CreateEvent{
+	jobId := queue.JobId{ProjectID: e.Event.ProjectID, ResourceID: e.Event.UID}.ReplayJobId()
+	createReplayEvent := task.CreateEvent{
+		JobID: jobId,
 		Event: e.Event,
 	}
-	createEvent.Event.AcknowledgedAt = null.TimeFrom(time.Now())
+	createReplayEvent.Event.AcknowledgedAt = null.TimeFrom(time.Now())
 
-	eventByte, err := msgpack.EncodeMsgPack(createEvent)
+	eventByte, err := msgpack.EncodeMsgPack(createReplayEvent)
 	if err != nil {
 		return &ServiceError{ErrMsg: err.Error()}
 	}
@@ -37,12 +38,10 @@ func (e *ReplayEventService) Run(ctx context.Context) error {
 	if util.IsStringEmpty(e.Event.UID) || util.IsStringEmpty(e.Event.ProjectID) {
 		return &ServiceError{ErrMsg: "missing event or project id"}
 	}
-	jobId := fmt.Sprintf("replay:%s:%s", e.Event.ProjectID, e.Event.UID)
 
 	job := &queue.Job{
 		ID:      jobId,
 		Payload: eventByte,
-		Delay:   0,
 	}
 
 	err = e.Queue.Write(convoy.CreateEventProcessor, convoy.CreateEventQueue, job)
diff --git a/web/ui/dashboard/src/scss/_font.scss b/web/ui/dashboard/src/scss/_font.scss
index 542f1151..8b9e5d1a 100644
--- a/web/ui/dashboard/src/scss/_font.scss
+++ b/web/ui/dashboard/src/scss/_font.scss
@@ -446,3 +446,36 @@
 	font-weight: normal;
 	src: local('Menlo Regular'), url('../assets/fonts/Menlo-Regular.woff') format('woff');
 }
+
+/* Open Sauce Sans Font Faces */
+@font-face {
+	font-family: 'Open Sauce Sans';
+	font-style: normal;
+	font-weight: 400;
+	font-display: swap;
+	src: url('../assets/fonts/OpenSauceSans-Regular.ttf') format('truetype');
+}
+
+@font-face {
+	font-family: 'Open Sauce Sans';
+	font-style: normal;
+	font-weight: 500;
+	font-display: swap;
+	src: url('../assets/fonts/OpenSauceSans-Medium.ttf') format('truetype');
+}
+
+@font-face {
+	font-family: 'Open Sauce Sans';
+	font-style: normal;
+	font-weight: 600;
+	font-display: swap;
+	src: url('../assets/fonts/OpenSauceSans-SemiBold.ttf') format('truetype');
+}
+
+@font-face {
+	font-family: 'Open Sauce Sans';
+	font-style: normal;
+	font-weight: 700;
+	font-display: swap;
+	src: url('../assets/fonts/OpenSauceSans-Bold.ttf') format('truetype');
+}
diff --git a/web/ui/dashboard/src/styles.scss b/web/ui/dashboard/src/styles.scss
index f7b77a40..d5fdc81e 100644
--- a/web/ui/dashboard/src/styles.scss
+++ b/web/ui/dashboard/src/styles.scss
@@ -1,38 +1,5 @@
 @import './scss/font';
 
-/* Open Sauce Sans Font Faces */
-@font-face {
-  font-family: 'Open Sauce Sans';
-  font-style: normal;
-  font-weight: 400;
-  font-display: swap;
-  src: url('./assets/fonts/OpenSauceSans-Regular.ttf') format('truetype');
-}
-
-@font-face {
-  font-family: 'Open Sauce Sans';
-  font-style: normal;
-  font-weight: 500;
-  font-display: swap;
-  src: url('./assets/fonts/OpenSauceSans-Medium.ttf') format('truetype');
-}
-
-@font-face {
-  font-family: 'Open Sauce Sans';
-  font-style: normal;
-  font-weight: 600;
-  font-display: swap;
-  src: url('./assets/fonts/OpenSauceSans-SemiBold.ttf') format('truetype');
-}
-
-@font-face {
-  font-family: 'Open Sauce Sans';
-  font-style: normal;
-  font-weight: 700;
-  font-display: swap;
-  src: url('./assets/fonts/OpenSauceSans-Bold.ttf') format('truetype');
-}
-
 @import './scss/forms';
 @import './scss/prism';
 @import './scss/material-design-reset';
@@ -101,10 +68,17 @@
 	}
 
 	/* Apply font family to common text elements without overriding sizes */
+	/* Exclude Monaco Editor and code elements - they should use monospace fonts */
 	p, span, div, h1, h2, h3, h4, h5, h6, a, button, input, textarea, select, label, td, th, li {
 		font-family: 'Open Sauce Sans', -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif !important;
 	}
 
+	/* Monaco Editor should use monospace fonts, not Open Sauce Sans */
+	.monaco-editor,
+	.monaco-editor * {
+		font-family: 'Menlo Regular', ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, 'Liberation Mono', 'Courier New', monospace !important;
+	}
+
 	h1 {
 		@apply text-h1;
 	}
diff --git a/worker/consumer.go b/worker/consumer.go
index 395b1927..5b8cf601 100644
--- a/worker/consumer.go
+++ b/worker/consumer.go
@@ -14,11 +14,17 @@ import (
 	"github.com/frain-dev/convoy/worker/task"
 )
 
+// JobTracker is an optional interface for capturing job IDs during tests
+type JobTracker interface {
+	RecordJob(task *asynq.Task)
+}
+
 type Consumer struct {
-	queue queue.Queuer
-	mux   *asynq.ServeMux
-	srv   *asynq.Server
-	log   log.StdLogger
+	queue      queue.Queuer
+	mux        *asynq.ServeMux
+	srv        *asynq.Server
+	log        log.StdLogger
+	jobTracker JobTracker // optional, used only in E2E tests
 }
 
 func NewConsumer(ctx context.Context, consumerPoolSize int, q queue.Queuer, lo log.StdLogger, level log.Level) *Consumer {
@@ -84,8 +90,18 @@ func (c *Consumer) Stop() {
 	c.srv.Shutdown()
 }
 
+// SetJobTracker sets an optional job tracker for E2E tests
+func (c *Consumer) SetJobTracker(tracker JobTracker) {
+	c.jobTracker = tracker
+}
+
 func (c *Consumer) loggingMiddleware(h asynq.Handler, tel *telemetry.Telemetry) asynq.Handler {
 	return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
+		// Record job ID if tracker is set (for E2E tests)
+		if c.jobTracker != nil {
+			c.jobTracker.RecordJob(t)
+		}
+
 		traceProvider := otel.GetTracerProvider()
 		tracer := traceProvider.Tracer("asynq.workers")
 
diff --git a/worker/task/monitor_twitter_sources.go b/worker/task/monitor_twitter_sources.go
index 1940a2b9..06e6b4cb 100644
--- a/worker/task/monitor_twitter_sources.go
+++ b/worker/task/monitor_twitter_sources.go
@@ -8,6 +8,7 @@ import (
 	"github.com/go-redsync/redsync/v4"
 	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/hibiken/asynq"
+	"github.com/oklog/ulid/v2"
 
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/database"
@@ -115,8 +116,8 @@ func sendNotificationEmail(source datastore.Source, endpoint *datastore.Endpoint
 	}
 
 	job := &queue.Job{
+		ID:      ulid.Make().String(),
 		Payload: bytes,
-		Delay:   0,
 	}
 
 	err = q.Write(convoy.NotificationProcessor, convoy.DefaultQueue, job)
diff --git a/worker/task/process_batch_retry.go b/worker/task/process_batch_retry.go
index 35a9d12c..747fcdc0 100644
--- a/worker/task/process_batch_retry.go
+++ b/worker/task/process_batch_retry.go
@@ -154,8 +154,8 @@ func ProcessBatchRetry(
 				}
 
 				job := &queue.Job{
+					ID:      delivery.UID,
 					Payload: data,
-					Delay:   0,
 				}
 
 				err2 = queuer.Write(convoy.EventProcessor, convoy.EventQueue, job)
diff --git a/worker/task/process_event_channel.go b/worker/task/process_event_channel.go
index fb3e4ff4..06a3276e 100644
--- a/worker/task/process_event_channel.go
+++ b/worker/task/process_event_channel.go
@@ -66,7 +66,7 @@ func ProcessEventCreationByChannel(channel EventChannel, endpointRepo datastore.
 		var lastEvent, lastRunErrored, err = getLastTaskInfo(ctx, t, channel, eventQueue, eventRepo)
 		if err != nil {
 			log.WithError(err).Error("failed to get last task info")
-			// return err
+			return err
 		}
 
 		if lastEvent != nil && lastEvent.IsDuplicateEvent && !lastRunErrored {
@@ -117,7 +117,7 @@ func ProcessEventCreationByChannel(channel EventChannel, endpointRepo datastore.
 			return err
 		}
 
-		jobId := fmt.Sprintf("match_subs:%s:%s", event.ProjectID, event.UID)
+		jobId := queue.JobId{ProjectID: event.ProjectID, ResourceID: event.UID}.MatchSubsJobId()
 		job := &queue.Job{
 			ID:      jobId,
 			Payload: payload,
@@ -272,15 +272,6 @@ func getLastTaskInfo(ctx context.Context, t *asynq.Task, ch EventChannel, eventQ
 			return nil, false, err
 		}
 		jobID = broadcastEvent.JobID
-
-	case "default":
-		var createEvent CreateEvent
-		err := getTaskPayload(t, &createEvent)
-		if err != nil {
-			return nil, false, err
-		}
-		jobID = createEvent.JobID
-
 	case "dynamic":
 		var dynamicEvent models.DynamicEvent
 		err := getTaskPayload(t, &dynamicEvent)
@@ -288,7 +279,15 @@ func getLastTaskInfo(ctx context.Context, t *asynq.Task, ch EventChannel, eventQ
 			return nil, false, err
 		}
 		jobID = dynamicEvent.JobID
+	default:
+		var createEvent CreateEvent
+		err := getTaskPayload(t, &createEvent)
+		if err != nil {
+			return nil, false, err
+		}
+		jobID = createEvent.JobID
 	}
+
 	if util.IsStringEmpty(jobID) || !strings.Contains(jobID, ":") {
 		return nil, false, &EndpointError{Err: fmt.Errorf("cannot deduce jobID: %s", jobID)}
 	}
@@ -321,10 +320,10 @@ func getLastTaskInfo(ctx context.Context, t *asynq.Task, ch EventChannel, eventQ
 	return lastEvent, lastRunErrored, err
 }
 
-func getTaskPayload(t *asynq.Task, pojo interface{}) error {
-	err := msgpack.DecodeMsgPack(t.Payload(), &pojo)
+func getTaskPayload(t *asynq.Task, pogo interface{}) error {
+	err := msgpack.DecodeMsgPack(t.Payload(), &pogo)
 	if err != nil {
-		err := json.Unmarshal(t.Payload(), &pojo)
+		err = json.Unmarshal(t.Payload(), &pogo)
 		if err != nil {
 			return err
 		}
diff --git a/worker/task/process_event_creation.go b/worker/task/process_event_creation.go
index dc736d27..671b93ef 100644
--- a/worker/task/process_event_creation.go
+++ b/worker/task/process_event_creation.go
@@ -414,17 +414,11 @@ func writeEventDeliveriesToQueue(ctx context.Context, opts WriteEventDeliveriesT
 				Payload: data,
 			}
 
-			switch s.Type {
-			case datastore.SubscriptionTypeAPI:
+			if s.Type == datastore.SubscriptionTypeAPI {
 				err = opts.EventQueue.Write(convoy.EventProcessor, convoy.EventQueue, job)
 				if err != nil {
 					log.FromContext(ctx).WithError(err).Errorf("[asynq]: an error occurred sending event delivery to be dispatched")
 				}
-			case datastore.SubscriptionTypeCLI:
-				err = opts.EventQueue.Write(convoy.StreamCliEventsProcessor, convoy.StreamQueue, job)
-				if err != nil {
-					log.FromContext(ctx).WithError(err).Error("[asynq]: an error occurred sending event delivery to the stream queue")
-				}
 			}
 		}
 	}



Test patch:
diff --git a/.mise-tasks/test/_default b/.mise-tasks/test/_default
index d897d260..28f9a818 100755
--- a/.mise-tasks/test/_default
+++ b/.mise-tasks/test/_default
@@ -5,5 +5,7 @@
 
 set -e
 
-go run ./cmd config
-go test -tags integration -p 1 ./...
\ No newline at end of file
+go list ./... | grep -v '/e2e' | while IFS='' read -r line
+do
+  go test -p 1 "$line"
+done
\ No newline at end of file
diff --git a/.mise-tasks/test/e2e.sh b/.mise-tasks/test/e2e.sh
new file mode 100755
index 00000000..d6ab6915
--- /dev/null
+++ b/.mise-tasks/test/e2e.sh
@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+
+#MISE description="Run E2E tests for job ID generation refactor"
+#MISE dir="{{ config_root }}"
+
+set -e
+
+echo "🧪 Running Job ID E2E tests..."
+echo ""
+
+# Array of all job ID tests
+tests=(
+  "TestE2E_DirectEvent_AllSubscriptions"
+  "TestE2E_DirectEvent_MustMatchSubscription"
+  "TestE2E_FanOutEvent_AllSubscriptions"
+  "TestE2E_FanOutEvent_MustMatchSubscription"
+  "TestE2E_FormEndpoint_ContentType"
+  "TestE2E_FormEndpoint_WithCustomHeaders"
+  "TestE2E_OAuth2_SharedSecret"
+  "TestE2E_OAuth2_ClientAssertion"
+  "TestE2E_SingleEvent_JobID_Format"
+  "TestE2E_SingleEvent_JobID_Deduplication"
+  "TestE2E_FanoutEvent_JobID_Format"
+  "TestE2E_FanoutEvent_MultipleOwners"
+  "TestE2E_BroadcastEvent_JobID_Format"
+  "TestE2E_BroadcastEvent_AllSubscribers"
+  "TestE2E_DynamicEvent_JobID_Format"
+  "TestE2E_DynamicEvent_MultipleEventTypes"
+  "TestE2E_ReplayEvent_JobID_Format"
+  "TestE2E_ReplayEvent_MultipleReplays"
+)
+
+# Counter for passed tests
+passed=0
+failed=0
+
+# Run each test individually
+for test in "${tests[@]}"; do
+  echo "▶ Running: $test"
+  if go test -v ./e2e/... -run "^${test}$" -timeout 2m; then
+    echo "✅ $test passed"
+    ((passed++))
+  else
+    echo "❌ $test failed"
+    ((failed++))
+  fi
+  echo ""
+done
+
+# Summary
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+echo "📊 Test Summary:"
+echo "   Passed: $passed"
+echo "   Failed: $failed"
+echo "   Total:  $((passed + failed))"
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+
+if [ $failed -eq 0 ]; then
+  echo "✅ All Job ID E2E tests passed!"
+  exit 0
+else
+  echo "❌ Some tests failed"
+  exit 1
+fi
diff --git a/e2e/broadcast_event_jobid_e2e_test.go b/e2e/broadcast_event_jobid_e2e_test.go
new file mode 100644
index 00000000..c1148f3b
--- /dev/null
+++ b/e2e/broadcast_event_jobid_e2e_test.go
@@ -0,0 +1,168 @@
+package e2e
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/require"
+
+	convoy "github.com/frain-dev/convoy-go/v2"
+)
+
+func TestE2E_BroadcastEvent_JobID_Format(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	ownerID1 := env.Organisation.OwnerID + "_broadcast_1"
+	ownerID2 := env.Organisation.OwnerID + "_broadcast_2"
+
+	// Create two endpoints with different owners
+	endpoint1 := CreateEndpointViaSDK(t, c, 19970, ownerID1)
+	endpoint2 := CreateEndpointViaSDK(t, c, 19971, ownerID2)
+	t.Logf("Created endpoints: %s (owner1), %s (owner2)", endpoint1.UID, endpoint2.UID)
+
+	// Create subscriptions for both endpoints
+	CreateSubscriptionViaSDK(t, c, endpoint1.UID, []string{"broadcast.event"})
+	CreateSubscriptionViaSDK(t, c, endpoint2.UID, []string{"broadcast.event"})
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send broadcast event via API
+	traceID := "broadcast-jobid-" + ulid.Make().String()
+	broadcastPayload := map[string]interface{}{
+		"event_type": "broadcast.event",
+		"data": map[string]string{
+			"traceId": traceID,
+		},
+	}
+
+	body, err := json.Marshal(broadcastPayload)
+	require.NoError(t, err)
+
+	req, err := http.NewRequest("POST",
+		fmt.Sprintf("%s/api/v1/projects/%s/events/broadcast", env.ServerURL, env.Project.UID),
+		bytes.NewBuffer(body))
+	require.NoError(t, err)
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusCreated, resp.StatusCode)
+	resp.Body.Close()
+
+	t.Logf("Sent broadcast event with traceId: %s", traceID)
+
+	// Wait for job to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("broadcast:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one broadcast job ID")
+
+	t.Logf("Captured broadcast job IDs: %v", jobIDs)
+
+	// Verify job ID format: broadcast:{projectID}:{eventID}
+	VerifyJobIDFormat(t, jobIDs[0], "broadcast", env.Project.UID)
+
+	t.Logf("✅ Job ID format verified: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Broadcast event with correct job ID")
+}
+
+func TestE2E_BroadcastEvent_AllSubscribers(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	// Create three endpoints with different owners
+	ownerID1 := env.Organisation.OwnerID + "_bc_all_1"
+	ownerID2 := env.Organisation.OwnerID + "_bc_all_2"
+	ownerID3 := env.Organisation.OwnerID + "_bc_all_3"
+
+	endpoint1 := CreateEndpointViaSDK(t, c, 19972, ownerID1)
+	endpoint2 := CreateEndpointViaSDK(t, c, 19973, ownerID2)
+	endpoint3 := CreateEndpointViaSDK(t, c, 19974, ownerID3)
+	t.Logf("Created endpoints: %s, %s, %s", endpoint1.UID, endpoint2.UID, endpoint3.UID)
+
+	// Create subscriptions for all endpoints
+	CreateSubscriptionViaSDK(t, c, endpoint1.UID, []string{"global.broadcast"})
+	CreateSubscriptionViaSDK(t, c, endpoint2.UID, []string{"global.broadcast"})
+	CreateSubscriptionViaSDK(t, c, endpoint3.UID, []string{"global.broadcast"})
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send broadcast event
+	traceID := "broadcast-all-" + ulid.Make().String()
+	broadcastPayload := map[string]interface{}{
+		"event_type": "global.broadcast",
+		"data": map[string]string{
+			"traceId": traceID,
+			"message": "This should reach all subscribers",
+		},
+	}
+
+	body, err := json.Marshal(broadcastPayload)
+	require.NoError(t, err)
+
+	req, err := http.NewRequest("POST",
+		fmt.Sprintf("%s/api/v1/projects/%s/events/broadcast", env.ServerURL, env.Project.UID),
+		bytes.NewBuffer(body))
+	require.NoError(t, err)
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusCreated, resp.StatusCode)
+	resp.Body.Close()
+
+	t.Logf("Sent broadcast event with traceId: %s", traceID)
+
+	// Wait for job to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("broadcast:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one broadcast job ID")
+
+	// Verify job ID format
+	VerifyJobIDFormat(t, jobIDs[0], "broadcast", env.Project.UID)
+	t.Logf("✅ Broadcast job found: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Broadcast event reached all subscribers")
+}
diff --git a/e2e/dynamic_event_jobid_e2e_test.go b/e2e/dynamic_event_jobid_e2e_test.go
new file mode 100644
index 00000000..2f051ddd
--- /dev/null
+++ b/e2e/dynamic_event_jobid_e2e_test.go
@@ -0,0 +1,161 @@
+package e2e
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/require"
+)
+
+func TestE2E_DynamicEvent_JobID_Format(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	webhookURL := fmt.Sprintf("http://localhost:%d/webhook", 19980)
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send dynamic event via API (creates endpoint on the fly)
+	traceID := "dynamic-jobid-" + ulid.Make().String()
+	dynamicPayload := map[string]interface{}{
+		"url":        webhookURL,
+		"secret":     "dynamic-secret",
+		"event_type": "dynamic.event",
+		"data": map[string]string{
+			"traceId": traceID,
+			"message": "Dynamic event creates endpoint inline",
+		},
+	}
+
+	body, err := json.Marshal(dynamicPayload)
+	require.NoError(t, err)
+
+	req, err := http.NewRequest("POST",
+		fmt.Sprintf("%s/api/v1/projects/%s/events/dynamic", env.ServerURL, env.Project.UID),
+		bytes.NewBuffer(body))
+	require.NoError(t, err)
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusCreated, resp.StatusCode)
+	resp.Body.Close()
+
+	t.Logf("Sent dynamic event with traceId: %s", traceID)
+
+	// Wait for job to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("dynamic:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one dynamic job ID")
+
+	t.Logf("Captured dynamic job IDs: %v", jobIDs)
+
+	// Verify job ID format: dynamic:{projectID}:{eventID}
+	VerifyJobIDFormat(t, jobIDs[0], "dynamic", env.Project.UID)
+
+	t.Logf("✅ Job ID format verified: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Dynamic event with correct job ID")
+}
+
+func TestE2E_DynamicEvent_MultipleEventTypes(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	webhookURL1 := fmt.Sprintf("http://localhost:%d/webhook", 19981)
+	webhookURL2 := fmt.Sprintf("http://localhost:%d/webhook", 19982)
+	webhookURL3 := fmt.Sprintf("http://localhost:%d/webhook", 19983)
+
+	// Clear validator before sending events
+	validator.Clear()
+
+	// Send multiple dynamic events with different event types
+	traceID1 := "dynamic-multi-1-" + ulid.Make().String()
+	traceID2 := "dynamic-multi-2-" + ulid.Make().String()
+	traceID3 := "dynamic-multi-3-" + ulid.Make().String()
+
+	// Dynamic event 1
+	payload1 := map[string]interface{}{
+		"url":        webhookURL1,
+		"secret":     "dynamic-secret-1",
+		"event_type": "dynamic.type.one",
+		"data":       map[string]string{"traceId": traceID1},
+	}
+
+	// Dynamic event 2
+	payload2 := map[string]interface{}{
+		"url":        webhookURL2,
+		"secret":     "dynamic-secret-2",
+		"event_type": "dynamic.type.two",
+		"data":       map[string]string{"traceId": traceID2},
+	}
+
+	// Dynamic event 3
+	payload3 := map[string]interface{}{
+		"url":        webhookURL3,
+		"secret":     "dynamic-secret-3",
+		"event_type": "dynamic.type.four",
+		"data":       map[string]string{"traceId": traceID3},
+	}
+
+	// Send all events
+	for i, payload := range []map[string]interface{}{payload1, payload2, payload3} {
+		body, err := json.Marshal(payload)
+		require.NoError(t, err)
+
+		req, err := http.NewRequest("POST",
+			fmt.Sprintf("%s/api/v1/projects/%s/events/dynamic", env.ServerURL, env.Project.UID),
+			bytes.NewBuffer(body))
+		require.NoError(t, err)
+
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+		client := &http.Client{}
+		resp, err := client.Do(req)
+		require.NoError(t, err)
+		require.Equal(t, http.StatusCreated, resp.StatusCode)
+		resp.Body.Close()
+
+		t.Logf("Sent dynamic event %d", i+1)
+	}
+
+	// Wait for jobs to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("dynamic:")
+	t.Logf("Captured %d dynamic job(s)", len(jobIDs))
+	require.GreaterOrEqual(t, len(jobIDs), 3, "Should have captured at least 3 dynamic jobs")
+
+	t.Log("✅ E2E test passed: Multiple dynamic events processed correctly")
+}
diff --git a/e2e/fanout_event_jobid_e2e_test.go b/e2e/fanout_event_jobid_e2e_test.go
new file mode 100644
index 00000000..fe24381c
--- /dev/null
+++ b/e2e/fanout_event_jobid_e2e_test.go
@@ -0,0 +1,117 @@
+package e2e
+
+import (
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/require"
+
+	convoy "github.com/frain-dev/convoy-go/v2"
+)
+
+func TestE2E_FanoutEvent_JobID_Format(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	ownerID := env.Organisation.OwnerID + "_fanout_test"
+
+	// Create two endpoints with same owner (no need for webhook servers)
+	endpoint1 := CreateEndpointViaSDK(t, c, 19960, ownerID)
+	endpoint2 := CreateEndpointViaSDK(t, c, 19961, ownerID)
+	t.Logf("Created endpoints: %s, %s", endpoint1.UID, endpoint2.UID)
+
+	// Create subscriptions for both endpoints
+	CreateSubscriptionViaSDK(t, c, endpoint1.UID, []string{"fanout.event"})
+	CreateSubscriptionViaSDK(t, c, endpoint2.UID, []string{"fanout.event"})
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send fanout event
+	traceID := "fanout-jobid-" + ulid.Make().String()
+	SendFanoutEventViaSDK(t, c, ownerID, "fanout.event", traceID)
+
+	t.Logf("Sent fanout event with traceId: %s", traceID)
+
+	// Wait for job to be processed by our test worker
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("fanout:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one fanout job ID")
+
+	t.Logf("Captured fanout job IDs: %v", jobIDs)
+
+	// Verify job ID format: fanout:{projectID}:{eventUID}
+	VerifyJobIDFormat(t, jobIDs[0], "fanout", env.Project.UID)
+
+	t.Logf("✅ Job ID format verified: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Fanout event with correct job ID")
+}
+
+func TestE2E_FanoutEvent_MultipleOwners(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	ownerID1 := env.Organisation.OwnerID + "_fanout_owner1"
+	ownerID2 := env.Organisation.OwnerID + "_fanout_owner2"
+
+	// Create endpoints: 2 for owner1, 1 for owner2
+	endpoint1 := CreateEndpointViaSDK(t, c, 19962, ownerID1)
+	endpoint2 := CreateEndpointViaSDK(t, c, 19963, ownerID1)
+	endpoint3 := CreateEndpointViaSDK(t, c, 19964, ownerID2)
+	t.Logf("Created endpoints: %s (owner1), %s (owner1), %s (owner2)", endpoint1.UID, endpoint2.UID, endpoint3.UID)
+
+	// Create subscriptions for all endpoints
+	CreateSubscriptionViaSDK(t, c, endpoint1.UID, []string{"multi.fanout.event"})
+	CreateSubscriptionViaSDK(t, c, endpoint2.UID, []string{"multi.fanout.event"})
+	CreateSubscriptionViaSDK(t, c, endpoint3.UID, []string{"multi.fanout.event"})
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send fanout event to owner1 (should reach endpoint1 and endpoint2 only)
+	traceID := "fanout-multi-" + ulid.Make().String()
+	SendFanoutEventViaSDK(t, c, ownerID1, "multi.fanout.event", traceID)
+
+	t.Logf("Sent fanout event to owner1 with traceId: %s", traceID)
+
+	// Wait for job to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("fanout:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one fanout job ID")
+
+	// Verify job ID format
+	VerifyJobIDFormat(t, jobIDs[0], "fanout", env.Project.UID)
+	t.Logf("✅ Found fanout job: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Fanout event correctly targets owner's endpoints only")
+}
diff --git a/e2e/replay_event_jobid_e2e_test.go b/e2e/replay_event_jobid_e2e_test.go
new file mode 100644
index 00000000..218e6b05
--- /dev/null
+++ b/e2e/replay_event_jobid_e2e_test.go
@@ -0,0 +1,150 @@
+package e2e
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/require"
+
+	"github.com/frain-dev/convoy/api/testdb"
+	"github.com/frain-dev/convoy/datastore"
+)
+
+func TestE2E_ReplayEvent_JobID_Format(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Note: For replay tests, we need events to exist in the database.
+	// Since our test worker doesn't create Event records (it just validates job IDs),
+	// we'll create an event directly in the database using testdb helper
+
+	// First create a dummy endpoint (required by SeedEvent)
+	endpoint, err := testdb.SeedEndpoint(env.App.DB, env.Project, ulid.Make().String(), "", "", false, datastore.ActiveEndpointStatus)
+	require.NoError(t, err)
+
+	eventUID := ulid.Make().String()
+	event, err := testdb.SeedEvent(
+		env.App.DB,
+		endpoint,
+		env.Project.UID,
+		eventUID,
+		"replay.event",
+		"", // sourceID not needed
+		[]byte(`{"test": "data"}`),
+	)
+	require.NoError(t, err)
+	t.Logf("Created event in database: %s", event.UID)
+
+	// Clear validator before replay
+	validator.Clear()
+
+	// Replay the event
+	client := &http.Client{}
+	replayReq, err := http.NewRequest("PUT",
+		fmt.Sprintf("%s/api/v1/projects/%s/events/%s/replay", env.ServerURL, env.Project.UID, event.UID),
+		nil)
+	require.NoError(t, err)
+	replayReq.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+	replayResp, err := client.Do(replayReq)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, replayResp.StatusCode)
+	replayResp.Body.Close()
+
+	t.Logf("Replayed event: %s", event.UID)
+
+	// Wait for replay job to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("replay:")
+	t.Logf("Captured %d replay job(s)", len(jobIDs))
+
+	// Verify replay job ID format if captured
+	if len(jobIDs) > 0 {
+		VerifyJobIDFormat(t, jobIDs[0], "replay", env.Project.UID)
+		t.Logf("✅ Replay job ID format verified: %s", jobIDs[0])
+	}
+
+	t.Log("✅ E2E test passed: Replay event with correct job ID")
+}
+
+func TestE2E_ReplayEvent_MultipleReplays(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Note: For replay tests, we need events to exist in the database.
+	// Since our test worker doesn't create Event records (it just validates job IDs),
+	// we'll create an event directly in the database using testdb helper
+
+	// First create a dummy endpoint (required by SeedEvent)
+	endpoint, err := testdb.SeedEndpoint(env.App.DB, env.Project, ulid.Make().String(), "", "", false, datastore.ActiveEndpointStatus)
+	require.NoError(t, err)
+
+	eventUID := ulid.Make().String()
+	event, err := testdb.SeedEvent(
+		env.App.DB,
+		endpoint,
+		env.Project.UID,
+		eventUID,
+		"multi.replay.event",
+		"", // sourceID not needed
+		[]byte(`{"test": "data"}`),
+	)
+	require.NoError(t, err)
+	t.Logf("Created event in database: %s", event.UID)
+
+	// Clear validator before replays
+	validator.Clear()
+
+	// Replay the event twice
+	client := &http.Client{}
+	for i := 1; i <= 2; i++ {
+		replayReq, err := http.NewRequest("PUT",
+			fmt.Sprintf("%s/api/v1/projects/%s/events/%s/replay", env.ServerURL, env.Project.UID, event.UID),
+			nil)
+		require.NoError(t, err)
+		replayReq.Header.Set("Authorization", "Bearer "+env.APIKey)
+
+		replayResp, err := client.Do(replayReq)
+		require.NoError(t, err)
+		require.Equal(t, http.StatusOK, replayResp.StatusCode)
+		replayResp.Body.Close()
+
+		t.Logf("Replay %d completed for event: %s", i, event.UID)
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	// Wait for replay jobs to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("replay:")
+	t.Logf("Captured %d replay job(s)", len(jobIDs))
+
+	// We should have captured 2 replay jobs
+	require.GreaterOrEqual(t, len(jobIDs), 2, "Should have captured at least 2 replay jobs")
+
+	t.Log("✅ E2E test passed: Multiple replay events processed correctly")
+}
diff --git a/e2e/setup_test.go b/e2e/setup_test.go
index bc297d82..fa534081 100644
--- a/e2e/setup_test.go
+++ b/e2e/setup_test.go
@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"os"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -38,6 +39,8 @@ var (
 	infra *testenv.Environment
 	// testMutex ensures tests run sequentially to avoid resource conflicts
 	testMutex sync.Mutex
+	// portCounter ensures unique ports across tests (starts at 15000)
+	portCounter atomic.Uint32
 )
 
 // TestMain sets up the test environment
@@ -73,6 +76,7 @@ type E2ETestEnv struct {
 	Organisation *datastore.Organisation
 	User         *datastore.User
 	APIKey       string
+	JobTracker   *testenv.JobTracker
 	ctx          context.Context
 	cancelWorker context.CancelFunc
 	cancelServer context.CancelFunc
@@ -160,6 +164,9 @@ func SetupE2E(t *testing.T) *E2ETestEnv {
 	// Create licenser
 	licenser := noopLicenser.NewLicenser()
 
+	// Create job tracker for E2E tests to capture job IDs
+	jobTracker := testenv.NewJobTracker()
+
 	// Create cli.App with all dependencies
 	app := &cli.App{
 		Version:       "test",
@@ -171,6 +178,7 @@ func SetupE2E(t *testing.T) *E2ETestEnv {
 		Rate:          limiter,
 		Licenser:      licenser,
 		TracerBackend: tracer.NoOpBackend{},
+		JobTracker:    jobTracker,
 	}
 
 	// Seed test data (before starting server/worker)
@@ -195,8 +203,14 @@ func SetupE2E(t *testing.T) *E2ETestEnv {
 	require.NoError(t, err)
 
 	// Use a unique port for each test to avoid conflicts
-	// Use a wide range (10000-20000) to minimize collision probability
-	serverPort := uint32(10000 + (time.Now().UnixNano() % 10000))
+	// Use atomic counter to ensure no port collisions
+	nextPort := portCounter.Add(1)
+	if nextPort == 1 {
+		// First test - initialize to a safe starting port
+		portCounter.Store(15000)
+		nextPort = 15000
+	}
+	serverPort := nextPort
 	cfg.Server.HTTP.Port = serverPort
 
 	// Override global config with our test-specific settings
@@ -258,12 +272,195 @@ func SetupE2E(t *testing.T) *E2ETestEnv {
 		Organisation: org,
 		User:         user,
 		APIKey:       apiKey,
+		JobTracker:   jobTracker,
 		ctx:          workerCtx,
 		cancelWorker: cancelWorker,
 		cancelServer: cancelServer,
 	}
 }
 
+// SetupE2EWithoutWorker initializes E2E test environment with server but WITHOUT worker
+// This is useful for job ID tests where we use a custom test worker instead
+func SetupE2EWithoutWorker(t *testing.T) *E2ETestEnv {
+	t.Helper()
+
+	// Lock to ensure tests run sequentially (prevents resource conflicts)
+	testMutex.Lock()
+
+	ctx := context.Background()
+
+	// Create logger
+	logger := testenv.NewLogger(t)
+	logger.SetLevel(log.ErrorLevel)
+
+	// Reload config for each test to ensure clean state
+	err := config.LoadConfig("")
+	require.NoError(t, err)
+
+	cfg, err := config.Get()
+	require.NoError(t, err)
+
+	// Clone database for this test
+	conn, err := infra.CloneTestDatabase(t, "convoy")
+	require.NoError(t, err)
+
+	// Initialize database hooks
+	dbHooks := hooks.Init()
+	dbHooks.RegisterHook(datastore.EndpointCreated, func(ctx context.Context, data interface{}, changelog interface{}) {})
+
+	pg := postgres.NewFromConnection(conn)
+
+	// Create Redis client
+	rd, err := infra.NewRedisClient(t, 0)
+	require.NoError(t, err)
+
+	// Set up key manager
+	_ = os.Setenv("CONVOY_LOCAL_ENCRYPTION_KEY", "test-key")
+	km, err := keys.NewLocalKeyManager("test-key")
+	require.NoError(t, err)
+	if km.IsSet() {
+		_, err = km.GetCurrentKeyFromCache()
+		require.NoError(t, err)
+	}
+	err = keys.Set(km)
+	require.NoError(t, err)
+
+	// Load CA cert
+	err = config.LoadCaCert("", "")
+	require.NoError(t, err)
+
+	// Create rdb client for queue
+	redis, err := rdb.NewClient(cfg.Redis.BuildDsn())
+	require.NoError(t, err)
+
+	// Create cache
+	cache := rcache.NewRedisCacheFromClient(rd)
+
+	// Create queue
+	queueNames := map[string]int{
+		string(convoy.EventQueue):         3,
+		string(convoy.CreateEventQueue):   3,
+		string(convoy.EventWorkflowQueue): 3,
+		string(convoy.ScheduleQueue):      1,
+		string(convoy.DefaultQueue):       1,
+		string(convoy.StreamQueue):        1,
+		string(convoy.MetaEventQueue):     1,
+	}
+
+	queueOpts := queue.QueueOptions{
+		RedisClient:  redis,
+		Names:        queueNames,
+		RedisAddress: cfg.Redis.BuildDsn(),
+		Type:         string(config.RedisQueueProvider),
+	}
+
+	q := redisqueue.NewQueue(queueOpts)
+
+	// Create rate limiter
+	limiter := rlimiter.NewLimiterFromRedisClient(rd)
+
+	// Create licenser
+	licenser := noopLicenser.NewLicenser()
+
+	// Create cli.App with all dependencies
+	app := &cli.App{
+		Version:       "test",
+		DB:            pg,
+		Redis:         rd,
+		Queue:         q,
+		Logger:        logger,
+		Cache:         cache,
+		Rate:          limiter,
+		Licenser:      licenser,
+		TracerBackend: tracer.NoOpBackend{},
+	}
+
+	// Seed test data (before starting server)
+	user, err := testdb.SeedDefaultUser(pg)
+	require.NoError(t, err)
+
+	org, err := testdb.SeedDefaultOrganisation(pg, user)
+	require.NoError(t, err)
+
+	project, err := testdb.SeedDefaultProjectWithSSL(pg, org.UID, &datastore.SSLConfiguration{EnforceSecureEndpoints: false})
+	require.NoError(t, err)
+
+	role := auth.Role{
+		Type:    auth.RoleProjectAdmin,
+		Project: project.UID,
+	}
+	_, apiKey, err := testdb.SeedAPIKey(pg, role, "", "test", "", "")
+	require.NoError(t, err)
+
+	// Seed configuration (required by worker)
+	_, err = testdb.SeedConfiguration(pg)
+	require.NoError(t, err)
+
+	// Use a unique port for each test to avoid conflicts
+	// Use atomic counter to ensure no port collisions
+	nextPort := portCounter.Add(1)
+	if nextPort == 1 {
+		// First test - initialize to a safe starting port
+		portCounter.Store(15000)
+		nextPort = 15000
+	}
+	serverPort := nextPort
+	cfg.Server.HTTP.Port = serverPort
+
+	// Override global config with our test-specific settings
+	err = config.Override(&cfg)
+	require.NoError(t, err)
+
+	serverURL := fmt.Sprintf("http://localhost:%d", serverPort)
+
+	// Start HTTP Server using cmd function with context for cancellation
+	serverCtx, cancelServer := context.WithCancel(ctx)
+	go func() {
+		err := cmdserver.StartConvoyServer(app)
+		if err != nil {
+			logger.WithError(err).Error("Server error")
+		}
+	}()
+
+	// Wait for server to start
+	waitForServer(t, serverURL, 10*time.Second)
+
+	// NOTE: We do NOT start the real worker here
+	// Tests using this setup should start their own custom test worker
+
+	t.Logf("Test environment ready (without worker): %s (port: %d, project: %s)", t.Name(), serverPort, project.UID)
+
+	// Cleanup function
+	t.Cleanup(func() {
+		t.Logf("Cleaning up test: %s", t.Name())
+		cancelServer()
+		// Give server time to fully shut down
+		time.Sleep(2 * time.Second)
+
+		// Reset memory store to prevent "table already registered" errors in next test
+		memorystore.DefaultStore.Reset()
+
+		// Unlock mutex to allow next test to run
+		testMutex.Unlock()
+		t.Logf("Cleanup complete for test: %s", t.Name())
+	})
+
+	return &E2ETestEnv{
+		T:            t,
+		App:          app,
+		Config:       cfg,
+		ServerURL:    serverURL,
+		Project:      project,
+		Organisation: org,
+		User:         user,
+		APIKey:       apiKey,
+		JobTracker:   nil, // Not used for custom worker tests
+		ctx:          serverCtx,
+		cancelWorker: nil,
+		cancelServer: cancelServer,
+	}
+}
+
 // waitForServer waits for the server to be ready
 func waitForServer(t *testing.T, url string, timeout time.Duration) {
 	t.Helper()
diff --git a/e2e/single_event_jobid_e2e_test.go b/e2e/single_event_jobid_e2e_test.go
new file mode 100644
index 00000000..69b34f43
--- /dev/null
+++ b/e2e/single_event_jobid_e2e_test.go
@@ -0,0 +1,144 @@
+package e2e
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/require"
+
+	convoy "github.com/frain-dev/convoy-go/v2"
+)
+
+func TestE2E_SingleEvent_JobID_Format(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	// Setup test infrastructure
+	manifest := NewEventManifest()
+	done := make(chan bool, 1)
+	var counter atomic.Int64
+	counter.Store(1)
+
+	// Start mock webhook server
+	port := 19950
+	StartMockWebhookServer(t, manifest, done, &counter, port)
+
+	ownerID := env.Organisation.OwnerID + "_jobid_test"
+
+	// Create endpoint
+	endpoint := CreateEndpointViaSDK(t, c, port, ownerID)
+	t.Logf("Created endpoint: %s", endpoint.UID)
+
+	// Create subscription
+	subscription := CreateSubscriptionViaSDK(t, c, endpoint.UID, []string{"test.event"})
+	t.Logf("Created subscription: %s", subscription.UID)
+
+	// Clear validator before sending event
+	validator.Clear()
+
+	// Send event
+	traceID := "jobid-test-" + ulid.Make().String()
+	SendEventViaSDK(t, c, endpoint.UID, "test.event", traceID)
+
+	// Wait for job to be processed by our test worker
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("single:")
+	require.NotEmpty(t, jobIDs, "Should have captured at least one single event job ID")
+
+	t.Logf("Captured job IDs: %v", jobIDs)
+
+	// Verify job ID format: single:{projectID}:{eventUID}
+	VerifyJobIDFormat(t, jobIDs[0], "single", env.Project.UID)
+
+	t.Logf("✅ Job ID format verified: %s", jobIDs[0])
+
+	t.Log("✅ E2E test passed: Single event with correct job ID")
+}
+
+func TestE2E_SingleEvent_JobID_Deduplication(t *testing.T) {
+	// Setup E2E environment WITHOUT starting the real worker
+	env := SetupE2EWithoutWorker(t)
+
+	// Create job ID validator and custom test worker
+	validator := NewJobIDValidator(t)
+	testWorker := NewTestWorker(env.ctx, t, env.App.Queue, validator)
+	testWorker.Start()
+	defer testWorker.Stop()
+
+	// Give worker time to start
+	time.Sleep(500 * time.Millisecond)
+
+	// Create Convoy SDK client
+	c := convoy.New(env.ServerURL+"/api/v1", env.APIKey, env.Project.UID)
+
+	// Setup test infrastructure
+	manifest := NewEventManifest()
+	done := make(chan bool, 1)
+	var counter atomic.Int64
+	counter.Store(1) // Expect only 1 webhook despite sending twice
+
+	// Start mock webhook server
+	port := 19951
+	StartMockWebhookServer(t, manifest, done, &counter, port)
+
+	ownerID := env.Organisation.OwnerID + "_dedup_test"
+
+	// Create endpoint
+	endpoint := CreateEndpointViaSDK(t, c, port, ownerID)
+	t.Logf("Created endpoint: %s", endpoint.UID)
+
+	// Create subscription
+	subscription := CreateSubscriptionViaSDK(t, c, endpoint.UID, []string{"test.event"})
+	t.Logf("Created subscription: %s", subscription.UID)
+
+	// Clear validator before sending events
+	validator.Clear()
+
+	// Use same idempotency key for both events
+	idempotencyKey := ulid.Make().String()
+	traceID := "dedup-test-" + idempotencyKey
+
+	// Send same event twice with same idempotency key
+	SendEventWithIdempotencyKey(t, c, endpoint.UID, "test.event", traceID, idempotencyKey)
+	t.Logf("Sent event 1 with idempotency key: %s", idempotencyKey)
+
+	time.Sleep(500 * time.Millisecond)
+
+	SendEventWithIdempotencyKey(t, c, endpoint.UID, "test.event", traceID, idempotencyKey)
+	t.Logf("Sent event 2 with same idempotency key: %s", idempotencyKey)
+
+	// Wait for jobs to be processed
+	time.Sleep(2 * time.Second)
+
+	// Get captured job IDs from validator
+	jobIDs := validator.GetJobIDsWithPrefix("single:")
+	t.Logf("Captured %d single event job(s)", len(jobIDs))
+
+	// Note: Idempotency at the API level may not be fully implemented yet
+	// For now, just verify that jobs are processed and have valid formats
+	require.NotEmpty(t, jobIDs, "Should have captured at least one job")
+
+	// Verify all captured job IDs have correct format
+	for _, jobID := range jobIDs {
+		VerifyJobIDFormat(t, jobID, "single", env.Project.UID)
+	}
+
+	t.Logf("✅ Verified %d job ID(s) with correct format", len(jobIDs))
+	t.Log("✅ E2E test passed: Job IDs validated successfully")
+}
diff --git a/e2e/test_worker.go b/e2e/test_worker.go
new file mode 100644
index 00000000..1a0fd9d1
--- /dev/null
+++ b/e2e/test_worker.go
@@ -0,0 +1,144 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/hibiken/asynq"
+
+	"github.com/frain-dev/convoy"
+	"github.com/frain-dev/convoy/pkg/log"
+	"github.com/frain-dev/convoy/queue"
+	"github.com/frain-dev/convoy/testenv"
+	"github.com/frain-dev/convoy/worker"
+)
+
+// JobIDValidator stores captured job IDs for validation
+type JobIDValidator struct {
+	mu     sync.RWMutex
+	jobIDs []string
+	t      *testing.T
+}
+
+// NewJobIDValidator creates a new validator
+func NewJobIDValidator(t *testing.T) *JobIDValidator {
+	t.Helper()
+	return &JobIDValidator{
+		jobIDs: make([]string, 0),
+		t:      t,
+	}
+}
+
+// RecordJobID stores a job ID
+func (v *JobIDValidator) RecordJobID(jobID string) {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.jobIDs = append(v.jobIDs, jobID)
+	v.t.Logf("Captured job ID: %s", jobID)
+}
+
+// GetJobIDs returns all captured job IDs
+func (v *JobIDValidator) GetJobIDs() []string {
+	v.mu.RLock()
+	defer v.mu.RUnlock()
+	result := make([]string, len(v.jobIDs))
+	copy(result, v.jobIDs)
+	return result
+}
+
+// GetJobIDsWithPrefix returns job IDs matching a prefix
+func (v *JobIDValidator) GetJobIDsWithPrefix(prefix string) []string {
+	v.mu.RLock()
+	defer v.mu.RUnlock()
+	result := make([]string, 0)
+	for _, jobID := range v.jobIDs {
+		if strings.HasPrefix(jobID, prefix) {
+			result = append(result, jobID)
+		}
+	}
+	return result
+}
+
+// Clear removes all job IDs
+func (v *JobIDValidator) Clear() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.jobIDs = make([]string, 0)
+}
+
+// TestWorker is a custom worker for job ID validation tests
+type TestWorker struct {
+	consumer  *worker.Consumer
+	validator *JobIDValidator
+	ctx       context.Context
+	cancel    context.CancelFunc
+}
+
+// NewTestWorker creates a test worker that validates job IDs
+func NewTestWorker(ctx context.Context, t *testing.T, q queue.Queuer, validator *JobIDValidator) *TestWorker {
+	t.Helper()
+
+	logger := testenv.NewLogger(t)
+	logger.SetLevel(log.ErrorLevel)
+
+	workerCtx, cancel := context.WithCancel(ctx)
+
+	consumer := worker.NewConsumer(workerCtx, 3, q, logger, log.ErrorLevel)
+
+	tw := &TestWorker{
+		consumer:  consumer,
+		validator: validator,
+		ctx:       workerCtx,
+		cancel:    cancel,
+	}
+
+	// Register handlers for all event processor types
+	consumer.RegisterHandlers(convoy.CreateEventProcessor, tw.handleCreateEvent, nil)
+	consumer.RegisterHandlers(convoy.CreateBroadcastEventProcessor, tw.handleCreateEvent, nil)
+	consumer.RegisterHandlers(convoy.CreateDynamicEventProcessor, tw.handleCreateEvent, nil)
+
+	return tw
+}
+
+// handleCreateEvent validates the job ID and records it
+func (tw *TestWorker) handleCreateEvent(ctx context.Context, t *asynq.Task) error {
+	// Get the job ID from the task
+	jobID := t.ResultWriter().TaskID()
+
+	// Record the job ID for validation
+	tw.validator.RecordJobID(jobID)
+
+	// Validate basic job ID format
+	if !strings.Contains(jobID, ":") {
+		return fmt.Errorf("invalid job ID format: %s (expected format with colons)", jobID)
+	}
+
+	// Success - job ID is valid
+	return nil
+}
+
+// Start starts the test worker
+func (tw *TestWorker) Start() {
+	go func() {
+		tw.consumer.Start()
+	}()
+}
+
+// Stop stops the test worker
+func (tw *TestWorker) Stop() {
+	tw.cancel()
+	tw.consumer.Stop()
+}
+
+// VerifyJobIDFormat verifies that a job ID has the expected format
+func VerifyJobIDFormat(t *testing.T, jobID, expectedPrefix, projectID string) {
+	t.Helper()
+
+	expected := fmt.Sprintf("%s:%s:", expectedPrefix, projectID)
+	if !strings.HasPrefix(jobID, expected) {
+		t.Errorf("Job ID %s should start with %s", jobID, expected)
+	}
+}
diff --git a/internal/pkg/billing/client_test.go b/internal/pkg/billing/client_test.go
index ade06193..cdf6bda9 100644
--- a/internal/pkg/billing/client_test.go
+++ b/internal/pkg/billing/client_test.go
@@ -8,9 +8,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/frain-dev/convoy/config"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/frain-dev/convoy/config"
 )
 
 func setupTestClient(t *testing.T) (*HTTPClient, *httptest.Server) {
diff --git a/queue/redis/client_test.go b/queue/redis/client_test.go
index e02976e3..e7017104 100644
--- a/queue/redis/client_test.go
+++ b/queue/redis/client_test.go
@@ -96,14 +96,18 @@ func (s *RedisQueueIntegrationTestSuite) TestWrite() {
 
 	for _, tc := range tests {
 		s.Run(tc.name, func() {
+			id := ulid.Make().String()
+
 			eventDelivery := &datastore.EventDelivery{
 				UID: tc.eventDeliveryID,
 			}
+
 			job := &queue.Job{
+				ID:      id,
 				Payload: json.RawMessage(eventDelivery.UID),
 			}
 
-			taskName := convoy.TaskName(ulid.Make().String())
+			taskName := convoy.TaskName(id)
 			err := s.eventQueue.Write(taskName, convoy.EventQueue, job)
 			s.Require().NoError(err, "Failed to write to queue")
 		})
diff --git a/testenv/job_tracker.go b/testenv/job_tracker.go
new file mode 100644
index 00000000..8f1d5f1c
--- /dev/null
+++ b/testenv/job_tracker.go
@@ -0,0 +1,85 @@
+package testenv
+
+import (
+	"sync"
+
+	"github.com/hibiken/asynq"
+)
+
+// JobTracker captures job IDs during E2E tests for verification
+type JobTracker struct {
+	mu      sync.RWMutex
+	jobIDs  []string
+	enabled bool
+}
+
+// NewJobTracker creates a new job tracker for E2E tests
+func NewJobTracker() *JobTracker {
+	return &JobTracker{
+		jobIDs:  make([]string, 0),
+		enabled: true,
+	}
+}
+
+// RecordJob records a job ID from a task
+func (jt *JobTracker) RecordJob(task *asynq.Task) {
+	if !jt.enabled {
+		return
+	}
+
+	jt.mu.Lock()
+	defer jt.mu.Unlock()
+
+	// The task ID in Asynq is the job ID we're tracking
+	taskID := task.ResultWriter().TaskID()
+	jt.jobIDs = append(jt.jobIDs, taskID)
+}
+
+// GetJobIDs returns all recorded job IDs
+func (jt *JobTracker) GetJobIDs() []string {
+	jt.mu.RLock()
+	defer jt.mu.RUnlock()
+
+	// Return a copy to avoid race conditions
+	result := make([]string, len(jt.jobIDs))
+	copy(result, jt.jobIDs)
+	return result
+}
+
+// GetJobIDsWithPrefix returns all job IDs that start with the given prefix
+func (jt *JobTracker) GetJobIDsWithPrefix(prefix string) []string {
+	jt.mu.RLock()
+	defer jt.mu.RUnlock()
+
+	result := make([]string, 0)
+	for _, jobID := range jt.jobIDs {
+		if len(jobID) >= len(prefix) && jobID[:len(prefix)] == prefix {
+			result = append(result, jobID)
+		}
+	}
+	return result
+}
+
+// Clear removes all recorded job IDs
+func (jt *JobTracker) Clear() {
+	jt.mu.Lock()
+	defer jt.mu.Unlock()
+
+	jt.jobIDs = make([]string, 0)
+}
+
+// Disable stops recording job IDs
+func (jt *JobTracker) Disable() {
+	jt.mu.Lock()
+	defer jt.mu.Unlock()
+
+	jt.enabled = false
+}
+
+// Enable starts recording job IDs
+func (jt *JobTracker) Enable() {
+	jt.mu.Lock()
+	defer jt.mu.Unlock()
+
+	jt.enabled = true
+}
diff --git a/testenv/launch.go b/testenv/launch.go
index c0091438..f0eda452 100644
--- a/testenv/launch.go
+++ b/testenv/launch.go
@@ -4,14 +4,18 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"net/url"
+	"testing"
 	"time"
 
+	"github.com/hibiken/asynq"
 	"golang.org/x/sync/errgroup"
 )
 
 type Environment struct {
 	CloneTestDatabase PostgresDBCloneFunc
 	NewRedisClient    RedisClientFunc
+	NewQueueInspector QueueInspectorFunc
 }
 
 func Launch(ctx context.Context) (*Environment, func() error, error) {
@@ -25,9 +29,18 @@ func Launch(ctx context.Context) (*Environment, func() error, error) {
 		return nil, nil, fmt.Errorf("start redis container: %w", err)
 	}
 
+	// Get Redis address for queue inspector
+	redisAddr, err := rediscontainer.ConnectionString(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("get redis address: %w", err)
+	}
+
+	inspectorFactory := newQueueInspectorFactory(redisAddr)
+
 	res := &Environment{
 		CloneTestDatabase: cloner,
 		NewRedisClient:    rcFactory,
+		NewQueueInspector: inspectorFactory,
 	}
 
 	return res, func() error {
@@ -52,3 +65,20 @@ func Launch(ctx context.Context) (*Environment, func() error, error) {
 		return eg.Wait()
 	}, nil
 }
+
+// newQueueInspectorFactory creates a factory function for creating asynq inspectors
+func newQueueInspectorFactory(redisAddr string) QueueInspectorFunc {
+	return func(t *testing.T) *asynq.Inspector {
+		t.Helper()
+
+		// Parse the Redis connection string to extract host:port
+		// testcontainers returns "redis://localhost:port" but asynq expects "localhost:port"
+		uri, err := url.Parse(redisAddr)
+		if err != nil {
+			t.Fatalf("failed to parse redis connection string: %v", err)
+		}
+
+		redisOpt := asynq.RedisClientOpt{Addr: uri.Host}
+		return asynq.NewInspector(redisOpt)
+	}
+}
diff --git a/testenv/queue.go b/testenv/queue.go
new file mode 100644
index 00000000..b0641ea7
--- /dev/null
+++ b/testenv/queue.go
@@ -0,0 +1,129 @@
+package testenv
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/stretchr/testify/require"
+
+	"github.com/frain-dev/convoy"
+)
+
+// QueueInspectorFunc creates a new asynq inspector for testing
+type QueueInspectorFunc func(t *testing.T) *asynq.Inspector
+
+// GetTaskByID retrieves a task from the queue by its ID
+func GetTaskByID(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, taskID string) (*asynq.TaskInfo, error) {
+	t.Helper()
+	return inspector.GetTaskInfo(string(queueName), taskID)
+}
+
+// WaitForTask polls the queue until a task with the given ID appears or timeout
+func WaitForTask(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, taskID string, timeout time.Duration) *asynq.TaskInfo {
+	t.Helper()
+
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		taskInfo, err := inspector.GetTaskInfo(string(queueName), taskID)
+		if err == nil {
+			return taskInfo
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	t.Fatalf("Task %s not found in queue %s within %v", taskID, queueName, timeout)
+	return nil
+}
+
+// WaitForTaskWithPrefix polls the queue until a task with the given ID prefix appears or timeout
+func WaitForTaskWithPrefix(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, prefix string, timeout time.Duration) *asynq.TaskInfo {
+	t.Helper()
+
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		tasks, err := inspector.ListPendingTasks(string(queueName))
+		if err == nil {
+			for _, task := range tasks {
+				if strings.HasPrefix(task.ID, prefix) {
+					return task
+				}
+			}
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	t.Fatalf("Task with prefix %s not found in queue %s within %v", prefix, queueName, timeout)
+	return nil
+}
+
+// AssertTaskExists verifies a task exists in the queue with the expected ID
+func AssertTaskExists(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, expectedTaskID string) {
+	t.Helper()
+
+	taskInfo, err := inspector.GetTaskInfo(string(queueName), expectedTaskID)
+	require.NoError(t, err, "Task should exist in queue")
+	require.Equal(t, expectedTaskID, taskInfo.ID, "Task ID should match")
+}
+
+// AssertTaskNotDuplicated verifies that attempting to enqueue the same job ID doesn't create duplicates
+func AssertTaskNotDuplicated(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, taskID string) {
+	t.Helper()
+
+	// Get all tasks in queue
+	tasks, err := inspector.ListPendingTasks(string(queueName))
+	require.NoError(t, err)
+
+	count := 0
+	for _, task := range tasks {
+		if task.ID == taskID {
+			count++
+		}
+	}
+
+	require.LessOrEqual(t, count, 1, "Task should not be duplicated in queue")
+}
+
+// VerifyJobIDFormat checks if a job ID matches the expected pattern
+func VerifyJobIDFormat(t *testing.T, jobID, expectedPrefix, projectID string) {
+	t.Helper()
+
+	expected := fmt.Sprintf("%s:%s:", expectedPrefix, projectID)
+	require.True(t, strings.HasPrefix(jobID, expected),
+		"Job ID %s should start with %s", jobID, expected)
+}
+
+// FindTaskByPrefix finds the first task in the queue with the given prefix
+func FindTaskByPrefix(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, prefix string) *asynq.TaskInfo {
+	t.Helper()
+
+	tasks, err := inspector.ListPendingTasks(string(queueName))
+	require.NoError(t, err)
+
+	for _, task := range tasks {
+		if strings.HasPrefix(task.ID, prefix) {
+			return task
+		}
+	}
+
+	return nil
+}
+
+// CountTasksWithPrefix counts how many tasks have the given prefix
+func CountTasksWithPrefix(t *testing.T, inspector *asynq.Inspector, queueName convoy.QueueName, prefix string) int {
+	t.Helper()
+
+	tasks, err := inspector.ListPendingTasks(string(queueName))
+	require.NoError(t, err)
+
+	count := 0
+	for _, task := range tasks {
+		if strings.HasPrefix(task.ID, prefix) {
+			count++
+		}
+	}
+
+	return count
+}
diff --git a/testenv/testing.go b/testenv/testing.go
index 47c7c911..c14d9c40 100644
--- a/testenv/testing.go
+++ b/testenv/testing.go
@@ -7,6 +7,13 @@ import (
 	"github.com/frain-dev/convoy/pkg/log"
 )
 
+// TestingT is an interface that matches the testing.T and testing.B types
+type TestingT interface {
+	Helper()
+	Fatalf(format string, args ...interface{})
+	Logf(format string, args ...interface{})
+}
+
 func NewLogger(t *testing.T) *log.Logger {
 	t.Helper()
 	return log.NewLogger(os.Stderr)