Skip to content

CROSSLINK-182 Implement Supplier actions and state transit #357

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JanisSaldabols merged 6 commits into from
Dec 18, 2025
Merged

CROSSLINK-182 Implement Supplier actions and state transit #357

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c3016c5
CROSSLINK-182 Implement Supplier actions and state transit
JanisSaldabols Dec 9, 2025
2b5ce64
CROSSLINK-182 Rename action mapping file
JanisSaldabols Dec 11, 2025
8015aff
CROSSLINK-182 Fix typos
JanisSaldabols Dec 11, 2025
f48acb7
CROSSLINK-182 Fix failing test
JanisSaldabols Dec 11, 2025
f7c95df
CROSSLINK-182 Create type for patron request side, state and action
JanisSaldabols Dec 15, 2025
a23e9cf
ILLDEV-253 Rename actions
JanisSaldabols Dec 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 33 additions & 32 deletions broker/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,38 +57,39 @@ Note that for all modes, the broker attaches Directory information about the sup

Configuration is provided via environment variables:

| Name | Description | Default value |
|------------------------|-------------------------------------------------------------------------------------------|-------------------------------------------|
| `HTTP_PORT` | Server port | `8081` |
| `DB_TYPE` | Database type | `postgres` |
| `DB_USER` | Database user | `crosslink` |
| `DB_PASSWORD` | Database password | `crosslink` |
| `DB_HOST` | Database host | `localhost` |
| `DB_DATABASE` | Database name | `crosslink` |
| `DB_PORT` | Database port | `25432` |
| `LOG_LEVEL` | Log level: `ERROR`, `WARN`, `INFO`, `DEBUG` | `INFO` |
| `ENABLE_JSON_LOG` | Should JSON log format be enabled | `false` |
| `BROKER_MODE` | Default broker mode if not configured for a peer: `opaque` or `transparent` | `opaque` |
| `BROKER_SYMBOL` | Symbol for the broker when in the `opaque` mode | `ISIL:BROKER` |
| `REQ_AGENCY_INFO` | Should `request/requestingAgencyInfo` be populated from Directory | `true` |
| `SUPPLIER_INFO` | Should `request/supplierInfo` be populated from Directory | `true` |
| `RETURN_INFO` | Should `returnInfo` be populated from Directory for supplier `Loaned` message | `true` |
| `VENDOR_NOTE` | Should `note` field be prepended with `Vendor: {vendor}` text | `true` |
| `SUPPLIER_SYMBOL_NOTE` | Should `note` field be prepended with a `Supplier: {symbol}` text, `opaque` mode only | `true` |
| `OFFERED_COSTS` | Should `deliveryCosts` be transferred to `offeredCosts` for ReShare vendor requesters | `false` |
| `NOTE_FIELD_SEP` | Separator for fields (e.g. Vendor) prepended to the note | `, ` |
| `CLIENT_DELAY` | Delay duration for outgoing ISO18626 messages | `0ms` |
| `SHUTDOWN_DELAY` | Delay duration for graceful shutdown (in-flight connections) | `15s` |
| `MAX_MESSAGE_SIZE` | Max accepted ISO18626 message size | `100KB` |
| `HOLDINGS_ADAPTER` | Holdings lookup method: `mock` or `sru` | `mock` |
| `SRU_URL` | Comma separated list of URLs when `HOLDINGS_ADAPTER` is `sru` | `http://localhost:8081/sru` |
| `DIRECTORY_ADAPTER` | Directory lookup method: `mock` or `api` | `mock` |
| `DIRECTORY_API_URL` | Comma separated list of URLs when `DIRECTORY_ADAPTER` is `api` | `http://localhost:8081/directory/entries` |
| `PEER_REFRESH_INTERVAL`| Peer refresh interval (via Directory lookup) | `5m` |
| `MOCK_CLIENT_URL` | Mocked peer URLs value when `DIRECTORY_ADAPTER` is `mock` | `http://localhost:19083/iso18626` |
| `API_PAGE_SIZE` | Default value for the `limit` query parameter when paging the API | `10` |
| `TENANT_TO_SYMBOL` | Pattern to map tenant to `requesterSymbol` when accessing the API via Okapi, | (empty value) |
| | the `{tenant}` token is replaced by the `X-Okapi-Tenant` header value | |
| Name | Description | Default value |
|--------------------------|---------------------------------------------------------------------------------------|-------------------------------------------|
| `HTTP_PORT` | Server port | `8081` |
| `DB_TYPE` | Database type | `postgres` |
| `DB_USER` | Database user | `crosslink` |
| `DB_PASSWORD` | Database password | `crosslink` |
| `DB_HOST` | Database host | `localhost` |
| `DB_DATABASE` | Database name | `crosslink` |
| `DB_PORT` | Database port | `25432` |
| `LOG_LEVEL` | Log level: `ERROR`, `WARN`, `INFO`, `DEBUG` | `INFO` |
| `ENABLE_JSON_LOG` | Should JSON log format be enabled | `false` |
| `BROKER_MODE` | Default broker mode if not configured for a peer: `opaque` or `transparent` | `opaque` |
| `BROKER_SYMBOL` | Symbol for the broker when in the `opaque` mode | `ISIL:BROKER` |
| `REQ_AGENCY_INFO` | Should `request/requestingAgencyInfo` be populated from Directory | `true` |
| `SUPPLIER_INFO` | Should `request/supplierInfo` be populated from Directory | `true` |
| `RETURN_INFO` | Should `returnInfo` be populated from Directory for supplier `Loaned` message | `true` |
| `VENDOR_NOTE` | Should `note` field be prepended with `Vendor: {vendor}` text | `true` |
| `SUPPLIER_SYMBOL_NOTE` | Should `note` field be prepended with a `Supplier: {symbol}` text, `opaque` mode only | `true` |
| `OFFERED_COSTS` | Should `deliveryCosts` be transferred to `offeredCosts` for ReShare vendor requesters | `false` |
| `NOTE_FIELD_SEP` | Separator for fields (e.g. Vendor) prepended to the note | `, ` |
| `CLIENT_DELAY` | Delay duration for outgoing ISO18626 messages | `0ms` |
| `SHUTDOWN_DELAY` | Delay duration for graceful shutdown (in-flight connections) | `15s` |
| `MAX_MESSAGE_SIZE` | Max accepted ISO18626 message size | `100KB` |
| `HOLDINGS_ADAPTER` | Holdings lookup method: `mock` or `sru` | `mock` |
| `SRU_URL` | Comma separated list of URLs when `HOLDINGS_ADAPTER` is `sru` | `http://localhost:8081/sru` |
| `DIRECTORY_ADAPTER` | Directory lookup method: `mock` or `api` | `mock` |
| `DIRECTORY_API_URL` | Comma separated list of URLs when `DIRECTORY_ADAPTER` is `api` | `http://localhost:8081/directory/entries` |
| `PEER_REFRESH_INTERVAL` | Peer refresh interval (via Directory lookup) | `5m` |
| `MOCK_CLIENT_URL` | Mocked peer URLs value when `DIRECTORY_ADAPTER` is `mock` | `http://localhost:19083/iso18626` |
| `API_PAGE_SIZE` | Default value for the `limit` query parameter when paging the API | `10` |
| `TENANT_TO_SYMBOL` | Pattern to map tenant to `requesterSymbol` when accessing the API via Okapi, | (empty value) |
| | the `{tenant}` token is replaced by the `X-Okapi-Tenant` header value | |
| `SUPPLIER_PATRON_PATTERN`| Pattern used to create patron ID when receiving Request on supplier side | `%v_user` |

# Build

Expand Down
15 changes: 8 additions & 7 deletions broker/events/eventmodels.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package events

import (
pr_db "github.com/indexdata/crosslink/broker/patron_request/db"
"github.com/indexdata/crosslink/httpclient"
"github.com/indexdata/crosslink/iso18626"
)
Expand Down Expand Up @@ -61,13 +62,13 @@ type EventData struct {
}

type CommonEventData struct {
IncomingMessage *iso18626.ISO18626Message `json:"incomingMessage,omitempty"`
OutgoingMessage *iso18626.ISO18626Message `json:"outgoingMessage,omitempty"`
Problem *Problem `json:"problem,omitempty"`
HttpFailure *httpclient.HttpError `json:"httpFailure,omitempty"`
EventError *EventError `json:"eventError,omitempty"`
Note string `json:"note,omitempty"`
Action *string `json:"action,omitempty"`
IncomingMessage *iso18626.ISO18626Message `json:"incomingMessage,omitempty"`
OutgoingMessage *iso18626.ISO18626Message `json:"outgoingMessage,omitempty"`
Problem *Problem `json:"problem,omitempty"`
HttpFailure *httpclient.HttpError `json:"httpFailure,omitempty"`
EventError *EventError `json:"eventError,omitempty"`
Note string `json:"note,omitempty"`
Action *pr_db.PatronRequestAction `json:"action,omitempty"`
}

type EventError struct {
Expand Down
4 changes: 4 additions & 0 deletions broker/handler/iso18626-handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ var waitingReqs = map[string]RequestWait{}
type Iso18626HandlerInterface interface {
HandleRequest(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter)
HandleRequestingAgencyMessage(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter)
HandleSupplyingAgencyMessage(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter)
}

type Iso18626Handler struct {
Expand Down Expand Up @@ -528,6 +529,9 @@ func handleRequestingAgencyErrorWithNotice(ctx common.ExtendedContext, w http.Re
}
}

func (h *Iso18626Handler) HandleSupplyingAgencyMessage(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter) {
handleSupplyingAgencyMessage(ctx, illMessage, w, h.illRepo, h.eventBus)
}
func handleSupplyingAgencyMessage(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter, repo ill_db.IllRepo, eventBus events.EventBus) {
var requestingRequestId = illMessage.SupplyingAgencyMessage.Header.RequestingAgencyRequestId
if requestingRequestId == "" {
Expand Down
2 changes: 2 additions & 0 deletions broker/migrations/015_add_properties_to_patron_request.down.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@

ALTER TABLE patron_request DROP COLUMN requester_req_id;
2 changes: 2 additions & 0 deletions broker/migrations/015_add_properties_to_patron_request.up.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@

ALTER TABLE patron_request ADD COLUMN requester_req_id VARCHAR;
28 changes: 16 additions & 12 deletions broker/patron_request/api/api-handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,16 @@ import (
var waitingReqs = map[string]RequestWait{}

type PatronRequestApiHandler struct {
prRepo pr_db.PrRepo
eventBus events.EventBus
prRepo pr_db.PrRepo
eventBus events.EventBus
actionMappingService prservice.ActionMappingService
}

func NewApiHandler(prRepo pr_db.PrRepo, eventBus events.EventBus) PatronRequestApiHandler {
return PatronRequestApiHandler{
prRepo: prRepo,
eventBus: eventBus,
prRepo: prRepo,
eventBus: eventBus,
actionMappingService: prservice.ActionMappingService{},
}
}

Expand Down Expand Up @@ -68,8 +70,8 @@ func (a *PatronRequestApiHandler) PostPatronRequests(w http.ResponseWriter, r *h
addInternalError(ctx, w, err)
return
}

_, err = a.eventBus.CreateTask(pr.ID, events.EventNameInvokeAction, events.EventData{CommonEventData: events.CommonEventData{Action: &prservice.ActionValidate}}, events.EventDomainPatronRequest, nil)
action := prservice.BorrowerActionValidate
_, err = a.eventBus.CreateTask(pr.ID, events.EventNameInvokeAction, events.EventData{CommonEventData: events.CommonEventData{Action: &action}}, events.EventDomainPatronRequest, nil)
if err != nil {
addInternalError(ctx, w, err)
return
Expand Down Expand Up @@ -133,7 +135,8 @@ func (a *PatronRequestApiHandler) GetPatronRequestsIdActions(w http.ResponseWrit
return
}
}
writeJsonResponse(w, prservice.GetBorrowerActionsByState(pr.State))
actions := a.actionMappingService.GetActionMapping(pr).GetActionsForPatronRequest(pr)
writeJsonResponse(w, actions)
}

func (a *PatronRequestApiHandler) PostPatronRequestsIdAction(w http.ResponseWriter, r *http.Request, id string, params proapi.PostPatronRequestsIdActionParams) {
Expand All @@ -156,12 +159,13 @@ func (a *PatronRequestApiHandler) PostPatronRequestsIdAction(w http.ResponseWrit
return
}
}
if !prservice.IsBorrowerActionAvailable(pr.State, action.Action) {

if !a.actionMappingService.GetActionMapping(pr).IsActionAvailable(pr, pr_db.PatronRequestAction(action.Action)) {
addBadRequestError(ctx, w, errors.New("Action "+action.Action+" is not allowed for patron request "+id))
return
}

data := events.EventData{CommonEventData: events.CommonEventData{Action: &action.Action}}
eventAction := pr_db.PatronRequestAction(action.Action)
data := events.EventData{CommonEventData: events.CommonEventData{Action: &eventAction}}
if action.ActionParams != nil {
data.CustomData = *action.ActionParams
}
Expand Down Expand Up @@ -232,8 +236,8 @@ func toApiPatronRequest(request pr_db.PatronRequest) proapi.PatronRequest {
return proapi.PatronRequest{
ID: request.ID,
Timestamp: request.Timestamp.Time,
State: request.State,
Side: request.Side,
State: string(request.State),
Side: string(request.Side),
Patron: toString(request.Patron),
RequesterSymbol: toString(request.RequesterSymbol),
SupplierSymbol: toString(request.SupplierSymbol),
Expand Down
1 change: 1 addition & 0 deletions broker/patron_request/api/api-handler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ func TestGetPatronRequestsId(t *testing.T) {

type PrRepoError struct {
mock.Mock
pr_db.PgPrRepo
}

func (r *PrRepoError) WithTxFunc(ctx common.ExtendedContext, fn func(repo pr_db.PrRepo) error) error {
Expand Down
5 changes: 5 additions & 0 deletions broker/patron_request/db/prmodels.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
package pr_db

type PatronRequestState string
type PatronRequestSide string
type PatronRequestAction string
16 changes: 16 additions & 0 deletions broker/patron_request/db/prrepo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package pr_db
import (
"github.com/indexdata/crosslink/broker/common"
"github.com/indexdata/crosslink/broker/repo"
"github.com/jackc/pgx/v5/pgtype"
)

type PrRepo interface {
Expand All @@ -11,6 +12,7 @@ type PrRepo interface {
ListPatronRequests(ctx common.ExtendedContext) ([]PatronRequest, error)
SavePatronRequest(ctx common.ExtendedContext, params SavePatronRequestParams) (PatronRequest, error)
DeletePatronRequest(ctx common.ExtendedContext, id string) error
GetPatronRequestBySupplierSymbolAndRequesterReqId(ctx common.ExtendedContext, supplierSymbol string, requesterReId string) (PatronRequest, error)
}

type PgPrRepo struct {
Expand Down Expand Up @@ -54,3 +56,17 @@ func (r *PgPrRepo) SavePatronRequest(ctx common.ExtendedContext, params SavePatr
func (r *PgPrRepo) DeletePatronRequest(ctx common.ExtendedContext, id string) error {
return r.queries.DeletePatronRequest(ctx, r.GetConnOrTx(), id)
}

func (r *PgPrRepo) GetPatronRequestBySupplierSymbolAndRequesterReqId(ctx common.ExtendedContext, supplierSymbol string, requesterReId string) (PatronRequest, error) {
row, err := r.queries.GetPatronRequestBySupplierSymbolAndRequesterReqId(ctx, r.GetConnOrTx(), GetPatronRequestBySupplierSymbolAndRequesterReqIdParams{
SupplierSymbol: pgtype.Text{
String: supplierSymbol,
Valid: true,
},
RequesterReqID: pgtype.Text{
String: requesterReId,
Valid: true,
},
})
return row.PatronRequest, err
}
Loading
Loading