From ce6df3c5c38ad26e60f410a954cdfca003a71e60 Mon Sep 17 00:00:00 2001
From: Zach Hirschtritt <zachary.hirschtritt@klaviyo.com>
Date: Wed, 22 Jan 2025 11:02:37 -0500
Subject: [PATCH] fix: add aggregate type to subquery to utilize indexes
 (#9226)

# Which Problems Are Solved

The subquery of the notification requested and retry requested is
missing the aggregate_type filter that would allow it to utilize the
`es_projection` or `active_instances_events` on the eventstore.events2
table.

# How the Problems Are Solved

Add additional filter on subquery. Final query:
```sql
SELECT <all the fields omitted> FROM eventstore.events2
WHERE
    instance_id = $1
    AND aggregate_type = $2
    AND event_type = $3
    AND created_at > $4
    AND aggregate_id NOT IN (
        SELECT aggregate_id
        FROM eventstore.events2
        WHERE
            aggregate_type = $5 <-- NB: previously missing
            AND event_type = ANY ($6)
            AND instance_id = $7
            AND created_at > $8
    )
ORDER BY "position", in_tx_order
LIMIT $9
FOR UPDATE SKIP LOCKED
```

# Additional Changes

# Additional Context

Co-authored-by: Livio Spring <livio.a@gmail.com>
(cherry picked from commit e4bbfcccc878eef185f912e414e03375224e173e)
---
 internal/notification/handlers/notification_worker.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/internal/notification/handlers/notification_worker.go b/internal/notification/handlers/notification_worker.go
index dbf11a72fd..9c00552acb 100644
--- a/internal/notification/handlers/notification_worker.go
+++ b/internal/notification/handlers/notification_worker.go
@@ -436,6 +436,7 @@ func (w *NotificationWorker) searchEvents(ctx context.Context, tx *sql.Tx, retry
 		Builder().
 		ExcludeAggregateIDs().
 		EventTypes(notification.RetryRequestedType, notification.CanceledType, notification.SentType).
+		AggregateTypes(notification.AggregateType).
 		Builder()
 	//nolint:staticcheck
 	return w.es.Filter(ctx, searchQuery)
@@ -454,6 +455,7 @@ func (w *NotificationWorker) searchRetryEvents(ctx context.Context, tx *sql.Tx)
 		Builder().
 		ExcludeAggregateIDs().
 		EventTypes(notification.CanceledType, notification.SentType).
+		AggregateTypes(notification.AggregateType).
 		Builder()
 	//nolint:staticcheck
 	events, err := w.es.Filter(ctx, searchQuery)