Index validation also for composite indexes

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] {

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

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)