CROSSLINK-266 Use CQL to search duplicates

Author: JanisSaldabols

broker/handler/iso18626-handler.go

@@ -11,9 +11,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/indexdata/cql-go/cqlbuilder"
 	"github.com/indexdata/crosslink/broker/catalog"
-	"github.com/indexdata/crosslink/broker/service"
 	"github.com/indexdata/crosslink/broker/shim"
+	"github.com/indexdata/crosslink/directory"
 
 	"github.com/indexdata/crosslink/broker/adapter"
 
@@ -34,6 +35,13 @@ var brokerSymbol = utils.GetEnv("BROKER_SYMBOL", "ISIL:BROKER")
 const HANDLER_COMP = "iso18626_handler"
 const ORIGINAL_INCOMING_MESSAGE = "originalIncomingMessage"
 
+var lookupQueryBuilder = utils.Must(catalog.NewQueryBuilderGen(&directory.QueryConfig{
+	Identifier: new("supplier_unique_record_id = {term}"),
+	Type:       new(directory.Cql),
+}))
+
+var queryTimeFormat = "2006-01-02 15:04:05"
+
 type ErrorValue string
 
 const (
@@ -201,23 +209,37 @@ func checkDuplicateRequest(ctx common.ExtendedContext, request *iso18626.Request
 		return nil
 	}
 
-	_, err := repo.FindDuplicateIllTransaction(ctx, ill_db.FindDuplicateIllTransactionParams{
-		RequesterSymbol: createPgText(requesterSymbol),
-		Hours:           service.ToInt32(*windowHours),
-		PatronID:        patronId,
-		Identifier:      lookupParams.Identifier,
-		Title:           lookupParams.Title,
-		ServiceType:     lookupParams.ServiceType,
-		Isbn:            lookupParams.Isbn,
-		Issn:            lookupParams.Issn,
-	})
+	cqlList, _, err := lookupQueryBuilder.Build(lookupParams)
+	if err != nil {
+		ctx.Logger().Warn("failed build lookup query", "error", err)
+		return nil
+	}
+
+	lookupCql := strings.Join(cqlList, " or ")
+	qb, err := cqlbuilder.NewQueryFromString("(" + lookupCql + ")")
+	if err != nil {
+		ctx.Logger().Warn("failed to build duplicate check query", "error", err)
+		return nil
+	}
+	formattedTime := time.Now().Add(-time.Duration(*windowHours) * time.Hour).Format(queryTimeFormat)
+	query, err := qb.And().Search("requester_symbol").Term(requesterSymbol).
+		And().Search("patron_id").Term(patronId).
+		And().Search("timestamp").Rel(">=").Term(formattedTime).
+		And().Search("service_type").Term(lookupParams.ServiceType).
+		Build()
+	if err != nil {
+		ctx.Logger().Warn("failed to build duplicate check query", "error", err)
+		return nil
+	}
+	cql := query.String()
+	trans, _, err := repo.ListIllTransactions(ctx, ill_db.ListIllTransactionsParams{Limit: 1, Offset: 0}, &cql, []string{requesterSymbol})
 	if err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return nil // no duplicate found
-		}
 		ctx.Logger().Warn("failed to check for duplicate requests, proceeding", "error", err)
 		return nil // fail open
 	}
+	if len(trans) == 0 {
+		return nil
+	}
 	return ErrDuplicateRequest
 }
 

broker/handler/iso18626-handler_test.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/indexdata/crosslink/broker/common"
@@ -160,16 +161,21 @@ func (r *MockIllRepositoryNoSelectedSupplier) GetSelectedSupplierForIllTransacti
 // configurable results for duplicate-check testing.
 type mockDuplicateCheckRepo struct {
 	mocks.MockIllRepositorySuccess
-	duplicateId string
-	err         error
-	called      bool
-	params      ill_db.FindDuplicateIllTransactionParams
+	duplicate bool
+	err       error
+	called    bool
+	cql       string
 }
 
-func (r *mockDuplicateCheckRepo) FindDuplicateIllTransaction(ctx common.ExtendedContext, params ill_db.FindDuplicateIllTransactionParams) (string, error) {
+func (r *mockDuplicateCheckRepo) ListIllTransactions(ctx common.ExtendedContext, params ill_db.ListIllTransactionsParams, cql *string, symbols []string) ([]ill_db.IllTransaction, int64, error) {
+	if cql != nil {
+		r.cql = *cql
+	}
 	r.called = true
-	r.params = params
-	return r.duplicateId, r.err
+	if r.duplicate {
+		return []ill_db.IllTransaction{{ID: "duplicate-id"}}, 1, nil
+	}
+	return []ill_db.IllTransaction{}, 0, r.err
 }
 
 func TestCheckDuplicateRequest(t *testing.T) {
@@ -207,12 +213,11 @@ func TestCheckDuplicateRequest(t *testing.T) {
 		name           string
 		request        *iso18626.Request
 		peer           ill_db.Peer
-		duplicateId    string
+		duplicate      bool
 		repoErr        error
 		wantErr        error
 		wantRepoCalled bool
 		wantPatronId   string
-		wantWindowHrs  int32
 		wantIdentifier string
 		wantIsbn       string
 		wantIssn       string
@@ -248,7 +253,6 @@ func TestCheckDuplicateRequest(t *testing.T) {
 			wantErr:        nil,
 			wantRepoCalled: true,
 			wantPatronId:   "patron-1",
-			wantWindowHrs:  1,
 			wantIdentifier: "rec-1",
 			wantTitle:      "Test Title",
 			wantSvcType:    "Loan",
@@ -261,7 +265,6 @@ func TestCheckDuplicateRequest(t *testing.T) {
 			wantErr:        nil,
 			wantRepoCalled: true,
 			wantPatronId:   "patron-1",
-			wantWindowHrs:  1,
 			wantIdentifier: "rec-1",
 			wantTitle:      "Test Title",
 			wantSvcType:    "Loan",
@@ -270,11 +273,10 @@ func TestCheckDuplicateRequest(t *testing.T) {
 			name:           "duplicate found - returns ErrDuplicateRequest",
 			request:        baseRequest,
 			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
-			duplicateId:    "existing-tx-id",
+			duplicate:      true,
 			wantErr:        ErrDuplicateRequest,
 			wantRepoCalled: true,
 			wantPatronId:   "patron-1",
-			wantWindowHrs:  1,
 			wantIdentifier: "rec-1",
 			wantTitle:      "Test Title",
 			wantSvcType:    "Loan",
@@ -290,6 +292,22 @@ func TestCheckDuplicateRequest(t *testing.T) {
 			wantErr:        nil,
 			wantRepoCalled: false,
 		},
+		{
+			name: "no service level - skips duplicate check (can't verify same patron)",
+			request: &iso18626.Request{
+				BibliographicInfo: iso18626.BibliographicInfo{
+					SupplierUniqueRecordId: "rec-1",
+					Title:                  "Test Title",
+				},
+				PatronInfo: &iso18626.PatronInfo{
+					PatronId: "patron-1",
+				},
+			},
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			repoErr:        pgx.ErrNoRows,
+			wantErr:        nil,
+			wantRepoCalled: false,
+		},
 		{
 			name:           "isbn passed as parameter to DB query",
 			request:        isbnRequest,
@@ -298,19 +316,28 @@ func TestCheckDuplicateRequest(t *testing.T) {
 			wantErr:        nil,
 			wantRepoCalled: true,
 			wantPatronId:   "patron-2",
-			wantWindowHrs:  1,
 			wantIsbn:       "978-1234",
 			wantSvcType:    "Copy",
 		},
 		{
 			name:           "duplicate found via isbn - returns ErrDuplicateRequest",
 			request:        isbnRequest,
 			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
-			duplicateId:    "existing-tx-isbn",
+			duplicate:      true,
 			wantErr:        ErrDuplicateRequest,
 			wantRepoCalled: true,
 			wantPatronId:   "patron-2",
-			wantWindowHrs:  1,
+			wantIsbn:       "978-1234",
+			wantSvcType:    "Copy",
+		},
+		{
+			name:           "no duplicate - returns nil",
+			request:        isbnRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			duplicate:      false,
+			wantErr:        nil,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-2",
 			wantIsbn:       "978-1234",
 			wantSvcType:    "Copy",
 		},
@@ -320,21 +347,19 @@ func TestCheckDuplicateRequest(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			appCtx := common.CreateExtCtxWithArgs(context.Background(), nil)
 			mockRepo := &mockDuplicateCheckRepo{
-				duplicateId: tt.duplicateId,
-				err:         tt.repoErr,
+				duplicate: tt.duplicate,
+				err:       tt.repoErr,
 			}
 			err := checkDuplicateRequest(appCtx, tt.request, mockRepo, "ISIL:REQ1", tt.peer)
 			assert.Equal(t, tt.wantErr, err)
 			assert.Equal(t, tt.wantRepoCalled, mockRepo.called)
 			if tt.wantRepoCalled {
-				assert.Equal(t, "ISIL:REQ1", mockRepo.params.RequesterSymbol.String)
-				assert.Equal(t, tt.wantPatronId, mockRepo.params.PatronID)
-				assert.Equal(t, tt.wantWindowHrs, mockRepo.params.Hours)
-				assert.Equal(t, tt.wantIdentifier, mockRepo.params.Identifier)
-				assert.Equal(t, tt.wantIsbn, mockRepo.params.Isbn)
-				assert.Equal(t, tt.wantIssn, mockRepo.params.Issn)
-				assert.Equal(t, tt.wantTitle, mockRepo.params.Title)
-				assert.Equal(t, tt.wantSvcType, mockRepo.params.ServiceType)
+				assert.True(t, tt.wantPatronId == "" || strings.Contains(mockRepo.cql, tt.wantPatronId))
+				assert.True(t, tt.wantIdentifier == "" || strings.Contains(mockRepo.cql, tt.wantIdentifier))
+				assert.True(t, tt.wantIsbn == "" || strings.Contains(mockRepo.cql, tt.wantIsbn))
+				assert.True(t, tt.wantIssn == "" || strings.Contains(mockRepo.cql, tt.wantIssn))
+				assert.True(t, tt.wantTitle == "" || strings.Contains(mockRepo.cql, tt.wantTitle))
+				assert.True(t, tt.wantSvcType == "" || strings.Contains(mockRepo.cql, tt.wantSvcType))
 			}
 		})
 	}

broker/ill_db/ill_cql.go

@@ -8,8 +8,12 @@ import (
 	"github.com/indexdata/cql-go/cql"
 	"github.com/indexdata/cql-go/cqlbuilder"
 	"github.com/indexdata/cql-go/pgcql"
+	pr_db "github.com/indexdata/crosslink/broker/patron_request/db"
+	"github.com/indexdata/go-utils/utils"
 )
 
+var LANGUAGE = utils.GetEnv("LANGUAGE", "english")
+
 func handleIllTransactionsQuery(cqlString string, noBaseArgs int) (pgcql.Query, error) {
 	def := pgcql.NewPgDefinition()
 
@@ -28,6 +32,24 @@ func handleIllTransactionsQuery(cqlString string, noBaseArgs int) (pgcql.Query,
 	f = pgcql.NewFieldString().WithExact()
 	def.AddField("last_requester_action", f)
 
+	def.AddField("isbn", pr_db.NewFieldTextArrayContains("bibliographic_item_identifiers(ill_transaction_data, 'ISBN')").WithFunction("norm_isxn"))
+	def.AddField("issn", pr_db.NewFieldTextArrayContains("bibliographic_item_identifiers(ill_transaction_data, 'ISSN')").WithFunction("norm_isxn"))
+
+	f = pgcql.NewFieldString().WithFullText(LANGUAGE).WithColumn("ill_transaction_data->'bibliographicInfo'->>'title'")
+	def.AddField("title", f)
+
+	nf := pgcql.NewFieldDate()
+	def.AddField("timestamp", nf)
+
+	f = pgcql.NewFieldString().WithLikeOps().WithColumn("ill_transaction_data->'patronInfo'->>'patronId'")
+	def.AddField("patron_id", f)
+
+	f = pgcql.NewFieldString().WithExact().WithColumn("ill_transaction_data->'bibliographicInfo'->>'supplierUniqueRecordId'")
+	def.AddField("supplier_unique_record_id", f)
+
+	f = pgcql.NewFieldString().WithExact().WithColumn("ill_transaction_data->'serviceInfo'->>'serviceType'")
+	def.AddField("service_type", f)
+
 	var parser cql.Parser
 	query, err := parser.Parse(cqlString)
 	if err != nil {

broker/ill_db/illrepo.go

@@ -50,7 +50,6 @@ type IllRepo interface {
 	GetExclusiveBranchSymbolsByPeerId(ctx common.ExtendedContext, peerId string) ([]BranchSymbol, error)
 	DeleteBranchSymbolByPeerId(ctx common.ExtendedContext, peerId string) error
 	CallArchiveIllTransactionByDateAndStatus(ctx common.ExtendedContext, toDate time.Time, statuses []string) error
-	FindDuplicateIllTransaction(ctx common.ExtendedContext, params FindDuplicateIllTransactionParams) (string, error)
 }
 
 type PgIllRepo struct {
@@ -478,10 +477,6 @@ func (r *PgIllRepo) CallArchiveIllTransactionByDateAndStatus(ctx common.Extended
 	return err
 }
 
-func (r *PgIllRepo) FindDuplicateIllTransaction(ctx common.ExtendedContext, params FindDuplicateIllTransactionParams) (string, error) {
-	return r.queries.FindDuplicateIllTransaction(ctx, r.GetConnOrTx(), params)
-}
-
 func (r *PgIllRepo) GetExclusiveBranchSymbolsByPeerId(ctx common.ExtendedContext, peerId string) ([]BranchSymbol, error) {
 	rows, err := r.queries.GetExclusiveBranchSymbolsByPeerId(ctx, r.GetConnOrTx(), peerId)
 	var symbols []BranchSymbol

broker/migrations/049_add_duplicate_search_index.down.sql

@@ -1 +1,13 @@
-DROP INDEX IF EXISTS idx_ill_transaction_requester_timestamp;
+DROP INDEX IF EXISTS idx_ill_transaction_requester_timestamp;
+
+DROP INDEX IF EXISTS idx_ill_transaction_isbn;
+
+DROP INDEX IF EXISTS idx_ill_transaction_issn;
+
+DROP INDEX idx_ill_transaction_title_tsv;
+
+DROP INDEX IF EXISTS idx_ill_transaction_supplier_unique_record_id;
+
+DROP INDEX IF EXISTS idx_ill_transaction_patron_id;
+
+DROP INDEX IF EXISTS idx_ill_transaction_service_type;

broker/migrations/049_add_duplicate_search_index.up.sql

@@ -1,3 +1,23 @@
 CREATE INDEX IF NOT EXISTS idx_ill_transaction_requester_timestamp
-    ON ill_transaction (requester_symbol, timestamp DESC)
-    INCLUDE (id);
+    ON ill_transaction (requester_symbol, timestamp)
+    INCLUDE (id);
+
+CREATE INDEX IF NOT EXISTS idx_ill_transaction_isbn
+    ON ill_transaction
+    USING gin (bibliographic_item_identifiers(ill_transaction_data, 'ISBN'));
+
+CREATE INDEX IF NOT EXISTS idx_ill_transaction_issn
+    ON ill_transaction
+    USING gin (bibliographic_item_identifiers(ill_transaction_data, 'ISSN'));
+
+CREATE INDEX idx_ill_transaction_title_tsv
+    ON ill_transaction USING GIN (to_tsvector('simple', ill_transaction_data->'bibliographicInfo'->>'title'));
+
+CREATE INDEX IF NOT EXISTS idx_ill_transaction_supplier_unique_record_id
+    ON ill_transaction ((ill_transaction_data->'bibliographicInfo'->>'supplierUniqueRecordId'));
+
+CREATE INDEX IF NOT EXISTS idx_ill_transaction_patron_id
+    ON ill_transaction ((ill_transaction_data->'patronInfo'->>'patronId'));
+
+CREATE INDEX IF NOT EXISTS idx_ill_transaction_service_type
+    ON ill_transaction ((ill_transaction_data->'serviceInfo'->>'serviceType'));

broker/sqlc/ill_query.sql

@@ -234,30 +234,3 @@ SELECT archive_ill_transaction_by_date_and_status($1, $2);
 SELECT sqlc.embed(branch_symbol)
 FROM branch_symbol b
 WHERE b.peer_id = $1 AND b.symbol_value not in (SELECT s.symbol_value FROM symbol s);
-
--- name: FindDuplicateIllTransaction :one
-SELECT ill_transaction.id
-FROM ill_transaction
-WHERE requester_symbol = $1
-  AND timestamp > NOW() - make_interval(hours => sqlc.arg(hours)::int4)
-  AND COALESCE(ill_transaction_data->'patronInfo'->>'patronId', '')                    = sqlc.arg(patron_id)::text
-  AND COALESCE(ill_transaction_data->'bibliographicInfo'->>'supplierUniqueRecordId', '') = sqlc.arg(identifier)::text
-  AND COALESCE(ill_transaction_data->'bibliographicInfo'->>'title', '')                = sqlc.arg(title)::text
-  AND COALESCE(ill_transaction_data->'serviceInfo'->>'serviceType', '')                = sqlc.arg(service_type)::text
-  AND COALESCE((
-    SELECT elem->>'bibliographicItemIdentifier'
-    FROM jsonb_array_elements(
-      COALESCE(ill_transaction_data->'bibliographicInfo'->'bibliographicItemId', '[]'::jsonb)
-    ) AS elem
-    WHERE elem->'bibliographicItemIdentifierCode'->>'#text' = 'ISBN'
-    LIMIT 1
-  ), '') = sqlc.arg(isbn)::text
-  AND COALESCE((
-    SELECT elem->>'bibliographicItemIdentifier'
-    FROM jsonb_array_elements(
-      COALESCE(ill_transaction_data->'bibliographicInfo'->'bibliographicItemId', '[]'::jsonb)
-    ) AS elem
-    WHERE elem->'bibliographicItemIdentifierCode'->>'#text' = 'ISSN'
-    LIMIT 1
-  ), '') = sqlc.arg(issn)::text
-LIMIT 1;