enforce uniqueness of origins on database level

Author: mgoetzegb

pkg/repository/migrations/000004_add_origins_table.up.sql

@@ -1,7 +1,7 @@
 CREATE TABLE notification_service.origins (
     "name"        TEXT NOT NULL,
-    "class"       TEXT NOT NULL,
-    "service_id"   TEXT NOT NULL
+    "class"       TEXT NOT NULL CONSTRAINT origins_class_unique UNIQUE,
+    "service_id"  TEXT NOT NULL
 );
 
 CREATE INDEX idx_origins_service_id ON notification_service.origins(service_id);

pkg/repository/originrepository/originRepository.go

@@ -10,9 +10,14 @@ import (
 	"fmt"
 
 	"github.com/greenbone/opensight-notification-service/pkg/entities"
+	"github.com/greenbone/opensight-notification-service/pkg/errs"
 	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
 )
 
+// see https://github.com/lib/pq/blob/3d613208bca2e74f2a20e04126ed30bcb5c4cc27/error.go#L78
+const pgErrCodeConflict = "23505"
+
 type OriginRepository struct {
 	client *sqlx.DB
 }
@@ -60,6 +65,12 @@ func (r *OriginRepository) UpsertOrigins(ctx context.Context, serviceID string,
 	if len(originRows) != 0 {
 		_, err = tx.NamedExec(createOriginsQuery, originRows)
 		if err != nil {
+			var pgErr *pq.Error
+			if errors.As(err, &pgErr) { // postgres specific error handling
+				if pgErr.Code == pgErrCodeConflict {
+					err = &errs.ErrConflict{Message: "duplicate origin class"}
+				}
+			}
 			return fmt.Errorf("could not insert origins: %w", err)
 		}
 	}

pkg/repository/originrepository/originRepository_test.go

@@ -6,6 +6,7 @@ package originrepository
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 
@@ -103,6 +104,23 @@ func Test_UpsertOrigins_ListOrigins(t *testing.T) {
 			},
 			wantOrigins: []entities.Origin{},
 		},
+		"error on duplicate origin classes": {
+			inputs: []input{
+				{
+					serviceID: "service1",
+					origins: []entities.Origin{
+						{Name: "origin1", Class: "classA"},
+					},
+				},
+				{
+					serviceID: "service2",
+					origins: []entities.Origin{
+						{Name: "origin1", Class: "classA"}, // class must be unique across all services
+					},
+				},
+			},
+			wantErr: true,
+		},
 		"error on empty serviceID": {
 			inputs: []input{
 				{
@@ -125,16 +143,19 @@ func Test_UpsertOrigins_ListOrigins(t *testing.T) {
 
 			ctx := context.Background()
 
+			var errs []error
 			for _, input := range tt.inputs {
 				err := repo.UpsertOrigins(ctx, input.serviceID, input.origins)
-				if tt.wantErr {
-					require.Error(t, err)
-				} else {
-					require.NoError(t, err)
-				}
+				errs = append(errs, err)
+			}
+			err = errors.Join(errs...)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
 			}
 
 			// if all operarions were successful, verify final state
+			require.NoError(t, err)
 			gotOrigins, err := repo.ListOrigins(ctx)
 			require.NoError(t, err)
 			assert.ElementsMatch(t, tt.wantOrigins, gotOrigins) // order so far not guaranteed or relevant
@@ -162,7 +183,9 @@ func Test_UpsertOrigins_Concurrency(t *testing.T) {
 		go func(val int) {
 			defer wg.Done()
 			err := repo.UpsertOrigins(context.Background(), serviceID, origins)
-			assert.NoError(t, err, "failed at iteration %d", val)
+			if err != nil {
+				t.Logf("Failed at iteration %d: %v", val, err)
+			}
 		}(i)
 	}
 	wg.Wait()
@@ -171,5 +194,5 @@ func Test_UpsertOrigins_Concurrency(t *testing.T) {
 	err = db.QueryRow("SELECT COUNT(*) FROM notification_service.origins WHERE service_id = $1", serviceID).Scan(&count)
 	require.NoError(t, err)
 
-	require.Equal(t, len(origins), count, "Data was duplicated due to race condition!")
+	require.Equal(t, 2, count, "Data was duplicated due to race condition!")
 }