feat: Add an API to allow for querying events under a subscription with additional filtering

docs/reference/config.md

@@ -1390,6 +1390,12 @@ nav_order: 2
 |batchSize|Default read ahead to enable for subscriptions that do not explicitly configure readahead|`int`|`50`
 |batchTimeout|Default batch timeout|`int`|`50ms`
 
+## subscription.events
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|maxScanLength|The maximum number of events a search for historical events matching a subscription will index from the database|`int`|`1000`
+
 ## subscription.retry
 
 |Key|Description|Type|Default Value|

internal/apiserver/route_get_subscription_events_filtered.go

@@ -0,0 +1,74 @@
+// Copyright © 2024 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apiserver
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"github.com/hyperledger/firefly-common/pkg/ffapi"
+	"github.com/hyperledger/firefly-common/pkg/i18n"
+	"github.com/hyperledger/firefly/internal/coremsgs"
+	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/hyperledger/firefly/pkg/database"
+)
+
+var getSubscriptionEventsFiltered = &ffapi.Route{
+	Name:   "getSubscriptionEventsFiltered",
+	Path:   "subscriptions/{subid}/events",
+	Method: http.MethodGet,
+	PathParams: []*ffapi.PathParam{
+		{Name: "subid", Description: coremsgs.APIParamsSubscriptionID},
+	},
+	QueryParams: []*ffapi.QueryParam{
+		{Name: "startsequence", IsBool: false, Description: coremsgs.APISubscriptionStartSequenceID},
+		{Name: "endsequence", IsBool: false, Description: coremsgs.APISubscriptionEndSequenceID},
+	},
+	FilterFactory:   database.EventQueryFactory,
+	Description:     coremsgs.APIEndpointsGetSubscriptionEventsFiltered,
+	JSONInputValue:  nil,
+	JSONOutputValue: func() interface{} { return []*core.Event{} },
+	JSONOutputCodes: []int{http.StatusOK},
+	Extensions: &coreExtensions{
+		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
+			subscription, _ := cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
+			var startSeq int
+			var endSeq int
+
+			if r.QP["startsequence"] != "" {
+				startSeq, err = strconv.Atoi(r.QP["startsequence"])
+				if err != nil {
+					return nil, i18n.NewError(cr.ctx, coremsgs.MsgSequenceIDDidNotParseToInt, fmt.Sprintf("startsequence: %s", r.QP["startsequence"]))
+				}
+			} else {
+				startSeq = -1
+			}
+
+			if r.QP["endsequence"] != "" {
+				endSeq, err = strconv.Atoi(r.QP["endsequence"])
+				if err != nil {
+					return nil, i18n.NewError(cr.ctx, coremsgs.MsgSequenceIDDidNotParseToInt, fmt.Sprintf("endsequence: %s", r.QP["endsequence"]))
+				}
+			} else {
+				endSeq = -1
+			}
+
+			return r.FilterResult(cr.or.GetSubscriptionEventsHistorical(cr.ctx, subscription, r.Filter, startSeq, endSeq))
+		},
+	},
+}

internal/apiserver/route_get_subscription_events_filtered_test.go

@@ -0,0 +1,85 @@
+// Copyright © 2024 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apiserver
+
+import (
+	"net/http/httptest"
+	"testing"
+
+	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestGetSubscriptionEventsFiltered(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?startsequence=100&endsequence=200", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+
+	o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
+		Return(&core.Subscription{}, nil)
+	o.On("GetSubscriptionEventsHistorical", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Return([]*core.EnrichedEvent{}, nil, nil)
+
+	r.ServeHTTP(res, req)
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
+
+func TestGetSubscriptionEventsFilteredStartSequenceIDDoesNotParse(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?startsequence=helloworld", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+	o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
+		Return(&core.Subscription{}, nil)
+
+	r.ServeHTTP(res, req)
+	assert.Equal(t, 400, res.Result().StatusCode)
+	assert.Contains(t, res.Body.String(), "helloworld")
+}
+
+func TestGetSubscriptionEventsFilteredEndSequenceIDDoesNotParse(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events?endsequence=helloworld", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+	o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
+		Return(&core.Subscription{}, nil)
+
+	r.ServeHTTP(res, req)
+	assert.Equal(t, 400, res.Result().StatusCode)
+	assert.Contains(t, res.Body.String(), "helloworld")
+}
+
+func TestGetSubscriptionEventsFilteredNoSequenceIDsProvided(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345/events", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+	o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
+		Return(&core.Subscription{}, nil)
+	o.On("GetSubscriptionEventsHistorical", mock.Anything, mock.Anything, mock.Anything, -1, -1).
+		Return([]*core.EnrichedEvent{}, nil, nil)
+
+	r.ServeHTTP(res, req)
+	assert.Equal(t, 200, res.Result().StatusCode)
+}

internal/apiserver/routes.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2024 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -110,6 +110,7 @@ var routes = append(
 		getStatusBatchManager,
 		getSubscriptionByID,
 		getSubscriptions,
+		getSubscriptionEventsFiltered,
 		getTokenAccountPools,
 		getTokenAccounts,
 		getTokenApprovals,

internal/apiserver/server.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2024 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -23,8 +23,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-
 	"github.com/gorilla/mux"
 	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
@@ -40,6 +38,7 @@ import (
 	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/namespace"
 	"github.com/hyperledger/firefly/internal/orchestrator"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 var (

internal/coreconfig/coreconfig.go

@@ -339,6 +339,8 @@ var (
 	SubscriptionsRetryMaxDelay = ffc("subscription.retry.maxDelay")
 	// SubscriptionsRetryFactor the backoff factor to use for retry of database operations
 	SubscriptionsRetryFactor = ffc("subscription.retry.factor")
+	// SubscriptionMaxHistoricalEventScanLength the maximum amount of historical events we scan for in the DB when indexing through old events against a subscription
+	SubscriptionMaxHistoricalEventScanLength = ffc("subscription.events.maxScanLength")
 	// TransactionWriterCount
 	TransactionWriterCount = ffc("transaction.writer.count")
 	// TransactionWriterBatchTimeout
@@ -459,6 +461,7 @@ func setDefaults() {
 	viper.SetDefault(string(SubscriptionsRetryInitialDelay), "250ms")
 	viper.SetDefault(string(SubscriptionsRetryMaxDelay), "30s")
 	viper.SetDefault(string(SubscriptionsRetryFactor), 2.0)
+	viper.SetDefault(string(SubscriptionMaxHistoricalEventScanLength), 1000)
 	viper.SetDefault(string(TransactionWriterBatchMaxTransactions), 100)
 	viper.SetDefault(string(TransactionWriterBatchTimeout), "10ms")
 	viper.SetDefault(string(TransactionWriterCount), 5)

internal/coremsgs/en_api_translations.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2024 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -133,6 +133,7 @@ var (
 	APIEndpointsGetWebSockets                   = ffm("api.endpoints.getStatusWebSockets", "Gets a list of the current WebSocket connections to this node")
 	APIEndpointsGetStatus                       = ffm("api.endpoints.getStatus", "Gets the status of this namespace")
 	APIEndpointsGetSubscriptionByID             = ffm("api.endpoints.getSubscriptionByID", "Gets a subscription by its ID")
+	APIEndpointsGetSubscriptionEventsFiltered   = ffm("api.endpoints.getSubscriptionEventsFiltered", "Gets a collection of events filtered by the subscription for further filtering")
 	APIEndpointsGetSubscriptions                = ffm("api.endpoints.getSubscriptions", "Gets a list of subscriptions")
 	APIEndpointsGetTokenAccountPools            = ffm("api.endpoints.getTokenAccountPools", "Gets a list of token pools that contain a given token account key")
 	APIEndpointsGetTokenAccounts                = ffm("api.endpoints.getTokenAccounts", "Gets a list of token accounts")
@@ -208,4 +209,7 @@ var (
 	APISmartContractDetails      = ffm("api.smartContractDetails", "Additional smart contract details")
 	APISmartContractDetailsKey   = ffm("api.smartContractDetailsKey", "Key")
 	APISmartContractDetailsValue = ffm("api.smartContractDetailsValue", "Value")
+
+	APISubscriptionStartSequenceID = ffm("api.startsequenceid", "The sequence ID in the raw event stream to start indexing through events from. Leave blank to start indexing from the most recent events")
+	APISubscriptionEndSequenceID   = ffm("api.endsequenceid", "The sequence ID in the raw event stream to stop indexing through events at. Leave blank to start indexing from the most recent events")
 )

internal/coremsgs/en_config_descriptions.go

@@ -383,9 +383,10 @@ var (
 	ConfigPluginSharedstorageIpfsGatewayURL      = ffc("config.plugins.sharedstorage[].ipfs.gateway.url", "The URL for the IPFS Gateway", urlStringType)
 	ConfigPluginSharedstorageIpfsGatewayProxyURL = ffc("config.plugins.sharedstorage[].ipfs.gateway.proxy.url", "Optional HTTP proxy server to use when connecting to the IPFS Gateway", urlStringType)
 
-	ConfigSubscriptionMax                  = ffc("config.subscription.max", "The maximum number of pre-defined subscriptions that can exist (note for high fan-out consider connecting a dedicated pub/sub broker to the dispatcher)", i18n.IntType)
-	ConfigSubscriptionDefaultsBatchSize    = ffc("config.subscription.defaults.batchSize", "Default read ahead to enable for subscriptions that do not explicitly configure readahead", i18n.IntType)
-	ConfigSubscriptionDefaultsBatchTimeout = ffc("config.subscription.defaults.batchTimeout", "Default batch timeout", i18n.IntType)
+	ConfigSubscriptionMax                          = ffc("config.subscription.max", "The maximum number of pre-defined subscriptions that can exist (note for high fan-out consider connecting a dedicated pub/sub broker to the dispatcher)", i18n.IntType)
+	ConfigSubscriptionDefaultsBatchSize            = ffc("config.subscription.defaults.batchSize", "Default read ahead to enable for subscriptions that do not explicitly configure readahead", i18n.IntType)
+	ConfigSubscriptionDefaultsBatchTimeout         = ffc("config.subscription.defaults.batchTimeout", "Default batch timeout", i18n.IntType)
+	ConfigSubscriptionMaxHistoricalEventScanLength = ffc("config.subscription.events.maxScanLength", "The maximum number of events a search for historical events matching a subscription will index from the database", i18n.IntType)
 
 	ConfigTokensName     = ffc("config.tokens[].name", "A name to identify this token plugin", i18n.StringType)
 	ConfigTokensPlugin   = ffc("config.tokens[].plugin", "The type of the token plugin to use", i18n.StringType)