Batch action owner restriction (#684)

* Add batch action runtime owner restriction

* Owner is optional

* Copilot

* Allow master tenant to create cross-owner batch actions

* Copilot

Author: jakub-id

broker/oapi/open-api.yaml

@@ -2516,7 +2516,7 @@ paths:
                 $ref: '#/components/schemas/Error'
     post:
       summary: Create a new batch action
-      description: Creates a batch action for one owner. Master access must provide `symbol`; Okapi access defaults to the tenant's primary symbol when omitted.
+      description: Creates a batch action. Master access without `symbol` creates an unrestricted action; Okapi access without `symbol` defaults to the tenant's primary symbol.
       tags:
         - scheduler-api
       parameters:

broker/scheduler/api/api_handler.go

@@ -123,7 +123,7 @@ func (h SchedulerApiHandler) PostBatchActions(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	owner, ok := h.resolveConcreteOwner(ctx, w, r, params.Symbol)
+	owner, ok := h.resolveBatchActionOwner(ctx, w, r, params.Symbol)
 	if !ok {
 		return
 	}
@@ -325,8 +325,9 @@ func (h SchedulerApiHandler) resolveOwnerScope(ctx common.ExtendedContext, w htt
 	return owners, true
 }
 
-// resolveConcreteOwner resolves the single owner required when creating a task.
-func (h SchedulerApiHandler) resolveConcreteOwner(ctx common.ExtendedContext, w http.ResponseWriter, r *http.Request, symbol *string) (string, bool) {
+// resolveBatchActionOwner resolves the owner for a new batch action. An empty
+// owner represents unrestricted master access.
+func (h SchedulerApiHandler) resolveBatchActionOwner(ctx common.ExtendedContext, w http.ResponseWriter, r *http.Request, symbol *string) (string, bool) {
 	t, err := h.tenantResolver.Resolve(ctx, r, symbol)
 	if err != nil {
 		brokerapi.AddBadRequestError(ctx, w, err)
@@ -337,10 +338,6 @@ func (h SchedulerApiHandler) resolveConcreteOwner(ctx common.ExtendedContext, w
 		brokerapi.AddBadRequestError(ctx, w, err)
 		return "", false
 	}
-	if owner == "" {
-		brokerapi.AddBadRequestError(ctx, w, errors.New("symbol must be specified when creating a batch action with master access"))
-		return "", false
-	}
 	return owner, true
 }
 

broker/scheduler/api/api_handler_test.go

@@ -409,17 +409,48 @@ func TestPostBatchActions_ActionParamsPersisted(t *testing.T) {
 	repo.AssertExpectations(t)
 }
 
-func TestPostBatchActions_MasterWithoutSymbolReturnsBadRequest(t *testing.T) {
+func TestPostBatchActions_MasterWithoutSymbolCreatesUnrestrictedAction(t *testing.T) {
 	repo := new(MockSchedRepo)
+	repo.On("SaveScheduledTask", mock.MatchedBy(func(p sched_db.SaveScheduledTaskParams) bool {
+		return p.Owner == "" &&
+			p.ActionData.BatchActionData != nil &&
+			p.ActionData.BatchActionData.Owner == ""
+	})).Return(saveScheduledTaskReturn, nil)
+
 	h := newHandler(repo)
-	req := newReq(http.MethodPost, `{"actionName":"email-pullslips","batchQuery":"title=test","schedule":"`+validRrule+`"}`)
+	req := newReq(http.MethodPost, `{"actionName":"request-aging","batchQuery":"title=test","schedule":"`+validRrule+`","actionParams":{"interval":"24h"}}`)
 	rr := httptest.NewRecorder()
 
 	h.PostBatchActions(rr, req, schedoapi.PostBatchActionsParams{})
 
-	assert.Equal(t, http.StatusBadRequest, rr.Code)
-	assert.Contains(t, rr.Body.String(), "symbol must be specified")
-	repo.AssertNotCalled(t, "SaveScheduledTask", mock.Anything)
+	assert.Equal(t, http.StatusCreated, rr.Code)
+	var resp schedoapi.BatchAction
+	assert.NoError(t, json.Unmarshal(rr.Body.Bytes(), &resp))
+	assert.Equal(t, schedoapi.RequestAging, resp.ActionName)
+	repo.AssertExpectations(t)
+}
+
+func TestPostBatchActions_OkapiWithoutSymbolUsesPrimaryMappedOwner(t *testing.T) {
+	const mappedOwner = "ISIL:DK-DIKU"
+	repo := new(MockSchedRepo)
+	repo.On("SaveScheduledTask", mock.MatchedBy(func(p sched_db.SaveScheduledTaskParams) bool {
+		return p.Owner == mappedOwner &&
+			p.ActionData.BatchActionData != nil &&
+			p.ActionData.BatchActionData.Owner == mappedOwner
+	})).Return(saveScheduledTaskReturn, nil)
+	resolver := tenant.NewResolver().
+		WithTenantToSymbol("ISIL:DK-{tenant}").
+		WithIllRepo(new(testmocks.MockIllRepositorySuccess))
+	h := NewSchedulerApiHandler(10, repo, nil, resolver)
+	req := httptest.NewRequest(http.MethodPost, "/broker/batch_actions",
+		strings.NewReader(`{"actionName":"request-aging","batchQuery":"title=test","schedule":"`+validRrule+`","actionParams":{"interval":"24h"}}`))
+	req.Header.Set(tenant.OkapiTenantHeader, "diku")
+	rr := httptest.NewRecorder()
+
+	h.PostBatchActions(rr, req, schedoapi.PostBatchActionsParams{})
+
+	assert.Equal(t, http.StatusCreated, rr.Code)
+	repo.AssertExpectations(t)
 }
 
 func TestPostBatchActions_MissingBody(t *testing.T) {

broker/scheduler/service/batch_action.go

@@ -1,6 +1,7 @@
 package sched_service
 
 import (
+	"fmt"
 	"strconv"
 	"time"
 
@@ -34,19 +35,67 @@ func (s *BatchActionService) BatchAction(ctx common.ExtendedContext, event event
 }
 func (s *BatchActionService) batchAction(ctx common.ExtendedContext, event events.Event) (events.EventStatus, *events.EventResult) {
 	ctx = ctx.WithArgs(ctx.LoggerArgs().WithComponent(BATCH_COMP))
-	if event.EventData.BatchActionData != nil {
-		switch event.EventData.BatchActionData.ActionName {
-		case string(schedoapi.EmailPullslips):
-			return s.emailSenderService.EmailPullslip(ctx, event)
-		case string(schedoapi.RequestAging):
-			return s.RequestAging(ctx, event)
-		default:
-			ctx.Logger().Error("unknown batch action", "actionName", event.EventData.BatchActionData.ActionName, "event", event)
-			return events.NewErrorResult("cannot process event", "unknown batch action")
-		}
+	if event.EventData.BatchActionData == nil {
+		ctx.Logger().Error("batch action data is empty", "eventId", event.ID)
+		return events.NewErrorResult("cannot process event", "batch action data is empty")
+	}
+
+	var action func(common.ExtendedContext, events.Event) (events.EventStatus, *events.EventResult)
+	switch event.EventData.BatchActionData.ActionName {
+	case string(schedoapi.EmailPullslips):
+		action = s.emailSenderService.EmailPullslip
+	case string(schedoapi.RequestAging):
+		action = s.RequestAging
+	default:
+		ctx.Logger().Error("unknown batch action",
+			"actionName", event.EventData.BatchActionData.ActionName,
+			"eventId", event.ID,
+			"taskId", event.EventData.BatchActionData.TaskId)
+		return events.NewErrorResult("cannot process event", "unknown batch action")
+	}
+
+	restrictedSelector, err := addBatchActionOwnerRestriction(
+		event.EventData.BatchActionData.Selector,
+		event.EventData.BatchActionData.Owner,
+	)
+	if err != nil {
+		return events.NewErrorResult("invalid batch action data", err.Error())
+	}
+
+	// Keep the event stored by the event bus unchanged while ensuring every
+	// action handler receives the owner-restricted selector.
+	batchActionData := *event.EventData.BatchActionData
+	batchActionData.Selector = restrictedSelector
+	event.EventData.BatchActionData = &batchActionData
+	return action(ctx, event)
+}
+
+func addBatchActionOwnerRestriction(selector string, owner string) (string, error) {
+	if selector == "" {
+		return "", fmt.Errorf("selector is empty")
+	}
+	if owner == "" {
+		return selector, nil
+	}
+
+	qb, err := cqlbuilder.NewQueryFromString(selector)
+	if err != nil {
+		return "", err
+	}
+	restrictedSelector, err := qb.And().
+		BeginClause().
+		Search("side").Term(string(prservice.SideLending)).
+		And().Search("supplier_symbol_exact").Term(owner).
+		Or().
+		BeginClause().Search("side").Term(string(prservice.SideBorrowing)).
+		And().Search("requester_symbol_exact").Term(owner).
+		EndClause().
+		EndClause().
+		Build()
+	if err != nil {
+		return "", err
 	}
-	ctx.Logger().Error("batch action data is empty", "event", event.ID)
-	return events.NewErrorResult("cannot process event", "batch action data is empty")
+	return restrictedSelector.String(), nil
 }
 
 func (s *BatchActionService) RequestAging(ctx common.ExtendedContext, event events.Event) (events.EventStatus, *events.EventResult) {
@@ -58,10 +107,6 @@ func (s *BatchActionService) RequestAging(ctx common.ExtendedContext, event even
 	if batchActionData.Selector == "" {
 		return events.NewErrorResult("cannot process event", "selector is empty")
 	}
-	if batchActionData.Owner == "" {
-		return events.NewErrorResult("cannot process event", "owner is empty")
-	}
-
 	intervalString, ok := event.EventData.CustomData["interval"].(string)
 	if !ok || intervalString == "" {
 		return events.NewErrorResult("cannot process event", "interval is missing or not a string")

broker/scheduler/service/batch_action_test.go

@@ -191,6 +191,37 @@ func TestBatchAction_RequestAgingDispatches(t *testing.T) {
 	assert.True(t, repo.listCalled)
 }
 
+func TestBatchAction_AddsOwnerRestrictionBeforeDispatch(t *testing.T) {
+	repo := &mockEmailPrRepo{}
+	svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, nil)
+	event := requestAgingEvent("state = REQ", map[string]any{"interval": "24h"})
+
+	status, _ := svc.batchAction(testCtx, event)
+
+	assert.Equal(t, events.EventStatusSuccess, status)
+	assert.Equal(t,
+		"(state = $3 AND ((side = $4 AND supplier_symbol = $5) OR (side = $6 AND requester_symbol = $7))) AND updated_at <= $8",
+		repo.gotQuery.GetWhereClause(),
+	)
+	// The restriction is added to the dispatched copy, not persisted into the
+	// scheduled event payload where repeated runs could accumulate predicates.
+	assert.Equal(t, "state = REQ", event.EventData.BatchActionData.Selector)
+}
+
+func TestBatchAction_MissingOwnerDispatchesWithoutRestriction(t *testing.T) {
+	repo := &mockEmailPrRepo{}
+	svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, nil)
+	event := requestAgingEvent("state = REQ", map[string]any{"interval": "24h"})
+	event.EventData.BatchActionData.Owner = ""
+
+	status, result := svc.batchAction(testCtx, event)
+
+	assert.Equal(t, events.EventStatusSuccess, status)
+	assert.Nil(t, result.EventError)
+	assert.True(t, repo.listCalled)
+	assert.Equal(t, "state = $3 AND updated_at <= $4", repo.gotQuery.GetWhereClause())
+}
+
 func TestRequestAging_ValidationErrors(t *testing.T) {
 	tests := []struct {
 		name              string
@@ -211,16 +242,6 @@ func TestRequestAging_ValidationErrors(t *testing.T) {
 			wantMsg:   "cannot process event",
 			wantCause: "selector is empty",
 		},
-		{
-			name: "empty owner",
-			event: func() events.Event {
-				event := requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"})
-				event.EventData.BatchActionData.Owner = ""
-				return event
-			}(),
-			wantMsg:   "cannot process event",
-			wantCause: "owner is empty",
-		},
 		{
 			name:      "missing custom data",
 			event:     requestAgingEvent("cql.allRecords=1", nil),