Add event_type to event_delivery table and use it to filter (frain-dev#1862)

Author: danvixent

.gitignore

@@ -15,7 +15,7 @@
 # vendor/
 .idea
 
-/api/dashboard/ui/build/*
+/api/ui/build/*
 !/api/ui/build/go_test_stub.txt
 /data/mongo/
 .DS_Store

api/api.go

@@ -183,9 +183,7 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 			})
 
 			r.HandleFunc("/*", handler.RedirectToProjects)
-
 		})
-
 	})
 
 	// Dashboard API.
@@ -222,12 +220,10 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 		})
 
 		uiRouter.Route("/organisations", func(orgRouter chi.Router) {
-
 			orgRouter.Post("/", handler.CreateOrganisation)
 			orgRouter.With(middleware.Pagination).Get("/", handler.GetOrganisationsPaged)
 
 			orgRouter.Route("/{orgID}", func(orgSubRouter chi.Router) {
-
 				orgSubRouter.Get("/", handler.GetOrganisation)
 				orgSubRouter.Put("/", handler.UpdateOrganisation)
 				orgSubRouter.Delete("/", handler.DeleteOrganisation)
@@ -253,7 +249,6 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 					projectRouter.Post("/", handler.CreateProject)
 
 					projectRouter.Route("/{projectID}", func(projectSubRouter chi.Router) {
-
 						projectSubRouter.Get("/", handler.GetProject)
 						projectSubRouter.Put("/", handler.UpdateProject)
 						projectSubRouter.Delete("/", handler.DeleteProject)
@@ -274,12 +269,10 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 								e.Put("/toggle_status", handler.ToggleEndpointStatus)
 								e.Put("/expire_secret", handler.ExpireSecret)
 								e.Put("/pause", handler.PauseEndpoint)
-
 							})
 						})
 
 						projectSubRouter.Route("/events", func(eventRouter chi.Router) {
-
 							eventRouter.Post("/", handler.CreateEndpointEvent)
 							eventRouter.Post("/fanout", handler.CreateEndpointFanoutEvent)
 							eventRouter.With(middleware.Pagination).Get("/", handler.GetEventsPaged)
@@ -320,7 +313,6 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 						})
 
 						projectSubRouter.Route("/sources", func(sourceRouter chi.Router) {
-
 							sourceRouter.Post("/", handler.CreateSource)
 							sourceRouter.Get("/{sourceID}", handler.GetSource)
 							sourceRouter.With(middleware.Pagination).Get("/", handler.LoadSourcesPaged)
@@ -351,7 +343,6 @@ func (a *ApplicationHandler) BuildRoutes() *chi.Mux {
 					})
 				})
 			})
-
 		})
 
 		uiRouter.Route("/configuration", func(configRouter chi.Router) {

database/postgres/event_delivery.go

@@ -37,8 +37,8 @@ var (
 
 const (
 	createEventDelivery = `
-    INSERT INTO convoy.event_deliveries (id,project_id,event_id,endpoint_id,device_id,subscription_id,headers,attempts,status,metadata,cli_metadata,description,url_query_params,idempotency_key,created_at,updated_at)
-    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16);
+    INSERT INTO convoy.event_deliveries (id,project_id,event_id,endpoint_id,device_id,subscription_id,headers,attempts,status,metadata,cli_metadata,description,url_query_params,idempotency_key,event_type,created_at,updated_at)
+    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17);
     `
 
 	baseFetchEventDelivery = `
@@ -48,6 +48,7 @@ const (
         COALESCE(ed.url_query_params, '') AS url_query_params,
         COALESCE(ed.idempotency_key, '') AS idempotency_key,
         ed.description,ed.created_at,ed.updated_at,
+        COALESCE(ed.event_type,'') AS "event_type",
         COALESCE(ed.device_id,'') AS "device_id",
         COALESCE(ed.endpoint_id,'') AS "endpoint_id",
         COALESCE(ep.id, '') AS "endpoint_metadata.id",
@@ -84,9 +85,7 @@ const (
 	    LIMIT :limit
 	)
 
-	SELECT * FROM event_deliveries
-	WHERE ("event_metadata.event_type" = :event_type OR :event_type = '')
-    ORDER BY id %s
+	SELECT * FROM event_deliveries ORDER BY id %s
 	`
 
 	baseEventDeliveryPagedBackward = `
@@ -99,32 +98,27 @@ const (
 		LIMIT :limit
 	)
 
-	SELECT * FROM event_deliveries
-	WHERE ("event_metadata.event_type" = :event_type OR :event_type = '')
-    ORDER BY id %s
+	SELECT * FROM event_deliveries ORDER BY id %s
 	`
 
 	fetchEventDeliveryByID = baseFetchEventDelivery + ` AND ed.id = $1 AND ed.project_id = $2`
 
 	baseEventDeliveryFilter = ` AND (ed.project_id = :project_id OR :project_id = '')
 	AND (ed.event_id = :event_id OR :event_id = '')
+    AND (ed.event_type = :event_type OR :event_type = '')
 	AND ed.created_at >= :start_date
 	AND ed.created_at <= :end_date
 	AND ed.deleted_at IS NULL`
 
 	countPrevEventDeliveries = `
-	WITH event_deliveries AS (
-        SELECT ed.id AS "id", ev.event_type AS "event_type"
-	    FROM convoy.event_deliveries ed
-    	LEFT JOIN convoy.events ev ON ed.event_id = ev.id
-	    WHERE ed.deleted_at IS NULL
-	    %s
-	    AND ed.id > :cursor
-        GROUP BY ed.id, ev.id
-        ORDER BY ed.id %s
-	)
-
-	SELECT COUNT(DISTINCT("id")) AS count FROM event_deliveries WHERE ("event_type" = :event_type OR :event_type = '')
+    SELECT COUNT(DISTINCT(ed.id))
+	FROM convoy.event_deliveries ed
+    LEFT JOIN convoy.events ev ON ed.event_id = ev.id
+	WHERE ed.deleted_at IS NULL
+	%s
+	AND ed.id > :cursor
+    GROUP BY ed.id, ev.id
+    ORDER BY ed.id %s
 	`
 
 	loadEventDeliveriesIntervals = `
@@ -153,6 +147,7 @@ const (
         COALESCE(ed.idempotency_key, '') AS idempotency_key,
         COALESCE(url_query_params, '') AS url_query_params,
         description,created_at,updated_at,
+        COALESCE(event_type,'') AS "event_type",
         COALESCE(device_id,'') AS "device_id",
         COALESCE(endpoint_id,'') AS "endpoint_id"
     FROM convoy.event_deliveries ed
@@ -165,6 +160,7 @@ const (
         COALESCE(idempotency_key, '') AS idempotency_key,
         COALESCE(url_query_params, '') AS url_query_params,
         description,created_at,updated_at,
+        COALESCE(event_type,'') AS "event_type",
         COALESCE(device_id,'') AS "device_id"
     FROM convoy.event_deliveries
 	WHERE status=$1 AND project_id = $2 AND device_id = $3
@@ -217,7 +213,7 @@ func (e *eventDeliveryRepo) CreateEventDelivery(ctx context.Context, delivery *d
 		ctx, createEventDelivery, delivery.UID, delivery.ProjectID,
 		delivery.EventID, endpointID, deviceID,
 		delivery.SubscriptionID, delivery.Headers, delivery.DeliveryAttempts, delivery.Status,
-		delivery.Metadata, delivery.CLIMetadata, delivery.Description, delivery.URLQueryParams, delivery.IdempotencyKey,
+		delivery.Metadata, delivery.CLIMetadata, delivery.Description, delivery.URLQueryParams, delivery.IdempotencyKey, delivery.EventType,
 		delivery.CreatedAt, delivery.UpdatedAt,
 	)
 	if err != nil {
@@ -580,6 +576,7 @@ func (e *eventDeliveryRepo) LoadEventDeliveriesPaged(ctx context.Context, projec
 			Headers:        ev.Headers,
 			URLQueryParams: ev.URLQueryParams,
 			Latency:        ev.Latency,
+			EventType:      ev.EventType,
 			Endpoint: &datastore.Endpoint{
 				UID:          ev.Endpoint.UID.ValueOrZero(),
 				ProjectID:    ev.Endpoint.ProjectID.ValueOrZero(),
@@ -824,6 +821,7 @@ type EventDeliveryPaginated struct {
 	URLQueryParams string                `json:"url_query_params" db:"url_query_params"`
 	IdempotencyKey string                `json:"idempotency_key" db:"idempotency_key"`
 	Latency        string                `json:"latency" db:"latency"`
+	EventType      datastore.EventType   `json:"event_type,omitempty" db:"event_type"`
 
 	Endpoint *EndpointMetadata `json:"endpoint_metadata,omitempty" db:"endpoint_metadata"`
 	Event    *EventMetadata    `json:"event_metadata,omitempty" db:"event_metadata"`

database/postgres/event_delivery_test.go

@@ -61,6 +61,7 @@ func generateEventDelivery(project *datastore.Project, endpoint *datastore.Endpo
 		EndpointID:     endpoint.UID,
 		DeviceID:       device.UID,
 		SubscriptionID: sub.UID,
+		EventType:      event.EventType,
 		Headers:        httpheader.HTTPHeader{"X-sig": []string{"3787 fmmfbf"}},
 		DeliveryAttempts: []datastore.DeliveryAttempt{
 			{UID: ulid.Make().String()},

datastore/models.go

@@ -892,6 +892,7 @@ type EventDelivery struct {
 	URLQueryParams string                `json:"url_query_params" db:"url_query_params"`
 	IdempotencyKey string                `json:"idempotency_key" db:"idempotency_key"`
 	Latency        string                `json:"latency" db:"latency"`
+	EventType      EventType             `json:"event_type,omitempty" db:"event_type"`
 
 	Endpoint *Endpoint `json:"endpoint_metadata,omitempty" db:"endpoint_metadata"`
 	Event    *Event    `json:"event_metadata,omitempty" db:"event_metadata"`

docs/docs.go

@@ -1,4 +1,6 @@
-// Code generated by swaggo/swag at 2024-01-03 08:49:59.960617 -0800 PST m=+2.346577667. DO NOT EDIT
+// Package docs GENERATED BY SWAG; DO NOT EDIT
+// This file was generated by swaggo/swag at
+// 2024-01-08 10:38:08.928262 +0000 GMT m=+2.574129084
 package docs
 
 import "github.com/swaggo/swag"
@@ -6656,6 +6658,9 @@ const docTemplate = `{
                 "event_metadata": {
                     "$ref": "#/definitions/datastore.Event"
                 },
+                "event_type": {
+                    "type": "string"
+                },
                 "headers": {
                     "$ref": "#/definitions/httpheader.HTTPHeader"
                 },

docs/swagger.json

@@ -6653,6 +6653,9 @@
                 "event_metadata": {
                     "$ref": "#/definitions/datastore.Event"
                 },
+                "event_type": {
+                    "type": "string"
+                },
                 "headers": {
                     "$ref": "#/definitions/httpheader.HTTPHeader"
                 },

docs/swagger.yaml

@@ -1038,6 +1038,8 @@ definitions:
         type: string
       event_metadata:
         $ref: '#/definitions/datastore.Event'
+      event_type:
+        type: string
       headers:
         $ref: '#/definitions/httpheader.HTTPHeader'
       idempotency_key:

docs/v3/openapi3.json

@@ -1501,6 +1501,9 @@
 					"event_metadata": {
 						"$ref": "#/components/schemas/datastore.Event"
 					},
+					"event_type": {
+						"type": "string"
+					},
 					"headers": {
 						"$ref": "#/components/schemas/httpheader.HTTPHeader"
 					},

docs/v3/openapi3.yaml

@@ -1039,6 +1039,8 @@ components:
           type: string
         event_metadata:
           $ref: '#/components/schemas/datastore.Event'
+        event_type:
+          type: string
         headers:
           $ref: '#/components/schemas/httpheader.HTTPHeader'
         idempotency_key:

sql/1704372039.sql

@@ -0,0 +1,7 @@
+-- +migrate Up
+ALTER TABLE convoy.event_deliveries ADD COLUMN IF NOT EXISTS event_type TEXT;
+CREATE INDEX IF NOT EXISTS event_deliveries_event_type_1 ON convoy.event_deliveries(event_type);
+
+-- +migrate Down
+ALTER TABLE convoy.event_deliveries DROP COLUMN IF EXISTS event_type;
+DROP INDEX IF EXISTS convoy.event_deliveries_event_type_1;

worker/task/process_event_creation.go

@@ -40,7 +40,8 @@ type CreateEvent struct {
 func ProcessEventCreation(
 	endpointRepo datastore.EndpointRepository, eventRepo datastore.EventRepository, projectRepo datastore.ProjectRepository,
 	eventDeliveryRepo datastore.EventDeliveryRepository, eventQueue queue.Queuer,
-	subRepo datastore.SubscriptionRepository, deviceRepo datastore.DeviceRepository) func(context.Context, *asynq.Task) error {
+	subRepo datastore.SubscriptionRepository, deviceRepo datastore.DeviceRepository,
+) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var createEvent CreateEvent
 		var event *datastore.Event
@@ -164,6 +165,7 @@ func ProcessEventCreation(
 			eventDelivery := &datastore.EventDelivery{
 				UID:              ulid.Make().String(),
 				SubscriptionID:   s.UID,
+				EventType:        event.EventType,
 				Metadata:         metadata,
 				ProjectID:        project.UID,
 				EventID:          event.UID,
@@ -227,7 +229,8 @@ func ProcessEventCreation(
 }
 
 func findSubscriptions(ctx context.Context, endpointRepo datastore.EndpointRepository,
-	subRepo datastore.SubscriptionRepository, project *datastore.Project, event *datastore.Event, shouldCreateSubscription bool) ([]datastore.Subscription, error) {
+	subRepo datastore.SubscriptionRepository, project *datastore.Project, event *datastore.Event, shouldCreateSubscription bool,
+) ([]datastore.Subscription, error) {
 	var subscriptions []datastore.Subscription
 	var err error
 
@@ -324,8 +327,8 @@ func matchSubscriptions(eventType string, subscriptions []datastore.Subscription
 }
 
 func getEventDeliveryStatus(ctx context.Context, subscription *datastore.Subscription, endpoint *datastore.Endpoint,
-	deviceRepo datastore.DeviceRepository) datastore.EventDeliveryStatus {
-
+	deviceRepo datastore.DeviceRepository,
+) datastore.EventDeliveryStatus {
 	switch subscription.Type {
 	case datastore.SubscriptionTypeAPI:
 		if endpoint.Status != datastore.ActiveEndpointStatus {
@@ -367,7 +370,8 @@ func generateSubscription(project *datastore.Project, endpoint *datastore.Endpoi
 }
 
 func buildEvent(ctx context.Context, eventRepo datastore.EventRepository, endpointRepo datastore.EndpointRepository,
-	eventParams *CreateEventTaskParams, project *datastore.Project) (*datastore.Event, error) {
+	eventParams *CreateEventTaskParams, project *datastore.Project,
+) (*datastore.Event, error) {
 	var isDuplicate bool
 	if !util.IsStringEmpty(eventParams.IdempotencyKey) {
 		events, err := eventRepo.FindEventsByIdempotencyKey(ctx, project.UID, eventParams.IdempotencyKey)
@@ -426,7 +430,8 @@ func buildEvent(ctx context.Context, eventRepo datastore.EventRepository, endpoi
 }
 
 func findEndpoints(ctx context.Context, endpointRepo datastore.EndpointRepository, newMessage *CreateEventTaskParams,
-	project *datastore.Project) ([]datastore.Endpoint, error) {
+	project *datastore.Project,
+) ([]datastore.Endpoint, error) {
 	var endpoints []datastore.Endpoint
 
 	if !util.IsStringEmpty(newMessage.EndpointID) {