CROSSLINK-266 Check incoming duplication messages (#660)

* CROSSLINK-266 Check incoming duplication messages

* CROSSLINK-266 Add duplicate action outcome

* CROSSLINK-266 Fix migration script

* CROSSLINK-266 Fix copilot comments

* CROSSLINK-266 Add duplicate test

* CROSSLINK-266 Use CQL to search duplicates

* CROSSLINK-266 Use exact match for title and patron id

Author: JanisSaldabols

broker/handler/iso18626-handler.go

@@ -11,8 +11,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/indexdata/cql-go/cqlbuilder"
 	"github.com/indexdata/crosslink/broker/catalog"
 	"github.com/indexdata/crosslink/broker/shim"
+	"github.com/indexdata/crosslink/directory"
 
 	"github.com/indexdata/crosslink/broker/adapter"
 
@@ -33,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 (
@@ -47,6 +56,7 @@ const (
 	InvalidAction             ErrorValue = "invalid action"
 	InvalidStatus             ErrorValue = "invalid status"
 	InvalidReason             ErrorValue = "invalid reason"
+	ReqIsDuplicate            ErrorValue = "duplicate request: an ILL request with the same patron, item, and service type was already submitted within the configured time window"
 )
 
 const PublicFailedToProcessReqMsg = "failed to process request"
@@ -64,6 +74,7 @@ var ErrReqAgencyNotFound = errors.New(string(ReqAgencyNotFound))
 var ErrInvalidAction = errors.New(string(InvalidAction))
 var ErrInvalidStatus = errors.New(string(InvalidStatus))
 var ErrInvalidReason = errors.New(string(InvalidReason))
+var ErrDuplicateRequest = errors.New(string(ReqIsDuplicate))
 
 var waitingReqs = map[string]RequestWait{}
 
@@ -139,6 +150,10 @@ func Iso18626PostHandler(repo ill_db.IllRepo, eventBus events.EventBus, dirAdapt
 }
 
 func handleNewRequest(ctx common.ExtendedContext, request *iso18626.Request, repo ill_db.IllRepo, requesterSymbol pgtype.Text, peers []ill_db.Peer) (string, error) {
+	if err := checkDuplicateRequest(ctx, request, repo, requesterSymbol.String, peers[0]); err != nil {
+		return "", err
+	}
+
 	supplierSymbol := createPgText(request.Header.SupplyingAgencyId.AgencyIdType.Text + ":" + request.Header.SupplyingAgencyId.AgencyIdValue)
 	requesterRequestId := createPgText(request.Header.RequestingAgencyRequestId)
 	supplierRequestId := createPgText(request.Header.SupplyingAgencyRequestId)
@@ -173,6 +188,61 @@ func handleNewRequest(ctx common.ExtendedContext, request *iso18626.Request, rep
 	return id, err
 }
 
+func checkDuplicateRequest(ctx common.ExtendedContext, request *iso18626.Request, repo ill_db.IllRepo, requesterSymbol string, peer ill_db.Peer) error {
+	windowHours := peer.CustomData.DuplicateCheckWindowHours
+	if windowHours == nil || *windowHours <= 0 {
+		return nil
+	}
+
+	patronId := ""
+	if request.PatronInfo != nil {
+		patronId = request.PatronInfo.PatronId
+	}
+
+	if patronId == "" {
+		return nil
+	}
+
+	lookupParams := catalog.LookupParamsFromBibliographicInfo(request.BibliographicInfo, request.ServiceInfo)
+
+	if lookupParams.ServiceType == "" {
+		return nil
+	}
+
+	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 {
+		ctx.Logger().Warn("failed to check for duplicate requests, proceeding", "error", err)
+		return nil // fail open
+	}
+	if len(trans) == 0 {
+		return nil
+	}
+	return ErrDuplicateRequest
+}
+
 func handleRetryRequest(ctx common.ExtendedContext, request *iso18626.Request, repo ill_db.IllRepo) (string, bool, error) {
 	// ServiceInfo already nil checked in handleIso18626Request
 	prevReqId := createPgText(request.ServiceInfo.RequestingAgencyPreviousRequestId)
@@ -268,6 +338,8 @@ func handleRequest(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Mess
 			handleRequestError(ctx, w, request, iso18626.TypeErrorTypeUnrecognisedDataValue, ReqIdAlreadyExists)
 		} else if errors.Is(err, ErrRetryNotPossible) {
 			handleRequestError(ctx, w, request, iso18626.TypeErrorTypeUnrecognisedDataValue, RetryNotPossible)
+		} else if errors.Is(err, ErrDuplicateRequest) {
+			handleRequestError(ctx, w, request, iso18626.TypeErrorTypeUnrecognisedDataValue, ReqIsDuplicate)
 		} else {
 			ctx.Logger().Error(InternalFailedToSaveTx, "error", err)
 			http.Error(w, PublicFailedToProcessReqMsg, http.StatusInternalServerError)

broker/handler/iso18626-handler_test.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/indexdata/crosslink/broker/common"
@@ -155,3 +156,211 @@ type MockIllRepositoryNoSelectedSupplier struct {
 func (r *MockIllRepositoryNoSelectedSupplier) GetSelectedSupplierForIllTransaction(ctx common.ExtendedContext, illTransId string) (ill_db.LocatedSupplier, error) {
 	return ill_db.LocatedSupplier{}, pgx.ErrNoRows
 }
+
+// mockDuplicateCheckRepo overrides FindDuplicateIllTransaction to return
+// configurable results for duplicate-check testing.
+type mockDuplicateCheckRepo struct {
+	mocks.MockIllRepositorySuccess
+	duplicate bool
+	err       error
+	called    bool
+	cql       string
+}
+
+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
+	if r.duplicate {
+		return []ill_db.IllTransaction{{ID: "duplicate-id"}}, 1, nil
+	}
+	return []ill_db.IllTransaction{}, 0, r.err
+}
+
+func TestCheckDuplicateRequest(t *testing.T) {
+	window1 := 1
+	window0 := 0
+	windowNeg := -1
+
+	baseRequest := &iso18626.Request{
+		BibliographicInfo: iso18626.BibliographicInfo{
+			SupplierUniqueRecordId: "rec-1",
+			Title:                  "Test Title",
+		},
+		ServiceInfo: &iso18626.ServiceInfo{
+			ServiceType: iso18626.TypeServiceTypeLoan,
+		},
+		PatronInfo: &iso18626.PatronInfo{
+			PatronId: "patron-1",
+		},
+	}
+
+	isbnRequest := &iso18626.Request{
+		BibliographicInfo: iso18626.BibliographicInfo{
+			BibliographicItemId: []iso18626.BibliographicItemId{
+				{
+					BibliographicItemIdentifier:     "978-1234",
+					BibliographicItemIdentifierCode: iso18626.TypeSchemeValuePair{Text: "ISBN"},
+				},
+			},
+		},
+		ServiceInfo: &iso18626.ServiceInfo{ServiceType: iso18626.TypeServiceTypeCopy},
+		PatronInfo:  &iso18626.PatronInfo{PatronId: "patron-2"},
+	}
+
+	tests := []struct {
+		name           string
+		request        *iso18626.Request
+		peer           ill_db.Peer
+		duplicate      bool
+		repoErr        error
+		wantErr        error
+		wantRepoCalled bool
+		wantPatronId   string
+		wantIdentifier string
+		wantIsbn       string
+		wantIssn       string
+		wantTitle      string
+		wantSvcType    string
+	}{
+		{
+			name:           "no DuplicateCheckWindowHours configured - skips check",
+			request:        baseRequest,
+			peer:           ill_db.Peer{},
+			wantErr:        nil,
+			wantRepoCalled: false,
+		},
+		{
+			name:           "window is zero - skips check",
+			request:        baseRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window0}},
+			wantErr:        nil,
+			wantRepoCalled: false,
+		},
+		{
+			name:           "window is negative - skips check",
+			request:        baseRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &windowNeg}},
+			wantErr:        nil,
+			wantRepoCalled: false,
+		},
+		{
+			name:           "db error - fails open, allows request through",
+			request:        baseRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			repoErr:        errors.New("db connection error"),
+			wantErr:        nil,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-1",
+			wantIdentifier: "rec-1",
+			wantTitle:      "Test Title",
+			wantSvcType:    "Loan",
+		},
+		{
+			name:           "no duplicate found (ErrNoRows) - not a duplicate",
+			request:        baseRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			repoErr:        pgx.ErrNoRows,
+			wantErr:        nil,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-1",
+			wantIdentifier: "rec-1",
+			wantTitle:      "Test Title",
+			wantSvcType:    "Loan",
+		},
+		{
+			name:           "duplicate found - returns ErrDuplicateRequest",
+			request:        baseRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			duplicate:      true,
+			wantErr:        ErrDuplicateRequest,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-1",
+			wantIdentifier: "rec-1",
+			wantTitle:      "Test Title",
+			wantSvcType:    "Loan",
+		},
+		{
+			name: "nil PatronInfo - skips duplicate check (can't verify same patron)",
+			request: &iso18626.Request{
+				BibliographicInfo: iso18626.BibliographicInfo{SupplierUniqueRecordId: "rec-1"},
+				ServiceInfo:       &iso18626.ServiceInfo{ServiceType: iso18626.TypeServiceTypeLoan},
+			},
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			repoErr:        pgx.ErrNoRows,
+			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,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			repoErr:        pgx.ErrNoRows,
+			wantErr:        nil,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-2",
+			wantIsbn:       "978-1234",
+			wantSvcType:    "Copy",
+		},
+		{
+			name:           "duplicate found via isbn - returns ErrDuplicateRequest",
+			request:        isbnRequest,
+			peer:           ill_db.Peer{CustomData: directory.Entry{DuplicateCheckWindowHours: &window1}},
+			duplicate:      true,
+			wantErr:        ErrDuplicateRequest,
+			wantRepoCalled: true,
+			wantPatronId:   "patron-2",
+			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",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			appCtx := common.CreateExtCtxWithArgs(context.Background(), nil)
+			mockRepo := &mockDuplicateCheckRepo{
+				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.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,6 +8,7 @@ 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"
 )
 
 func handleIllTransactionsQuery(cqlString string, noBaseArgs int) (pgcql.Query, error) {
@@ -28,6 +29,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().WithExact().WithLower().WithColumn("ill_transaction_data->'bibliographicInfo'->>'title'")
+	def.AddField("title", f)
+
+	nf := pgcql.NewFieldDate()
+	def.AddField("timestamp", nf)
+
+	f = pgcql.NewFieldString().WithExact().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/migrations/049_add_duplicate_search_index.down.sql

@@ -0,0 +1,13 @@
+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;