From 5464683ade7b1d0172c5a34f9d34a1c3049a2c46 Mon Sep 17 00:00:00 2001
From: Petr Stojanov <petrstojanov@gmail.com>
Date: Wed, 31 Jan 2024 10:08:13 +0100
Subject: [PATCH] Index validation also for composite indexes

---
 validator.go      |  4 +++-
 validator_test.go | 21 +++++++++++++++++++++
 x/schema/queue.go | 20 ++++++++++++++++++++
 3 files changed, 44 insertions(+), 1 deletion(-)

diff --git a/validator.go b/validator.go
index 6992ce1..b4af610 100644
--- a/validator.go
+++ b/validator.go
@@ -177,7 +177,9 @@ func unorderedEqual(first, second []string) bool {
 	}
 	exists := make(map[string]bool)
 	for _, value := range first {
-		exists[value] = true
+		for _, val := range strings.Split(value, ",") {
+			exists[val] = true
+		}
 	}
 	for _, value := range second {
 		if !exists[value] {
diff --git a/validator_test.go b/validator_test.go
index 46a4c66..5df9f7f 100644
--- a/validator_test.go
+++ b/validator_test.go
@@ -78,6 +78,27 @@ func TestValidator_ValidateIndexesCorrectSchema(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestValidator_ValidateIndexesCorrectSchema_CompositeIndexes(t *testing.T) {
+	// --- (1) ----
+	// Arrange
+	ctx := context.Background()
+	db := openDB(t)
+	queueName := fmt.Sprintf("TestQueue_%s", generateRandomString(10))
+
+	defer db.ExecContext(ctx, schema.GenerateDropTableQuery(queueName))
+
+	// Create the new queue
+	_, err := db.ExecContext(ctx, schema.GenerateCreateTableQueryCompositeIndex(queueName))
+	require.NoError(t, err)
+
+	// --- (2) ----
+	// Act: Validate queue
+	err = ValidateIndexes(db, queueName)
+
+	// Assert
+	require.NoError(t, err)
+}
+
 func TestValidator_ValidateIndexesIncorrectSchema(t *testing.T) {
 	// --- (1) ----
 	// Arrange
diff --git a/x/schema/queue.go b/x/schema/queue.go
index b9ef2a1..f752591 100644
--- a/x/schema/queue.go
+++ b/x/schema/queue.go
@@ -27,6 +27,26 @@ func GenerateCreateTableQuery(queueName string) string {
 	`, quotedTableName, quotedTableName[1:len(quotedTableName)-1])
 }
 
+// GenerateCreateTableQueryCompositeIndex returns the query for creating the queue table
+func GenerateCreateTableQueryCompositeIndex(queueName string) string {
+	quotedTableName := pg.QuoteIdentifier(queueName)
+	return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %[1]s
+	(
+		id             UUID        DEFAULT gen_random_uuid() NOT NULL PRIMARY KEY,
+		created_at     TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP NOT NULL,
+		started_at     TIMESTAMPTZ                           NULL,
+		locked_until   TIMESTAMPTZ                           NULL,
+		processed_at   TIMESTAMPTZ                           NULL,
+		consumed_count INTEGER     DEFAULT 0                 NOT NULL,
+		error_detail   TEXT                                  NULL,
+		payload        JSONB                                 NOT NULL,
+		metadata       JSONB                                 NOT NULL
+	);
+	CREATE INDEX IF NOT EXISTS "%[2]s_created_at_idx" ON %[1]s (created_at);
+	CREATE INDEX IF NOT EXISTS "%[2]s_processed_at_null_idx" ON %[1]s (consumed_count, processed_at) WHERE (processed_at IS NULL);
+	`, quotedTableName, quotedTableName[1:len(quotedTableName)-1])
+}
+
 // GenerateDropTableQuery returns a postgres query for dropping the queue table
 func GenerateDropTableQuery(queueName string) string {
 	quotedTableName := pg.QuoteIdentifier(queueName)