Repository: hyperledger/firefly
Pull request #1031 (2022-08-31 20:37:43)

update event docs
in a chain with #1028 

not currently published on my github pages due to hosting another doc changes. 


Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index eaf809ee9..df14e778b 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -3846,6 +3846,13 @@ paths:
         required: true
         schema:
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          example: "true"
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -13056,6 +13063,13 @@ paths:
         schema:
           example: default
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          example: "true"
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -22090,6 +22104,12 @@ paths:
         schema:
           example: default
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -22400,7 +22420,7 @@ paths:
       operationId: getTokenAccountPoolsNamespace
       parameters:
       - description: The key for the token account. The exact format may vary based
-          on the token connector use.
+          on the token connector use
         in: path
         name: key
         required: true
@@ -29661,6 +29681,12 @@ paths:
         required: true
         schema:
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -29964,7 +29990,7 @@ paths:
       operationId: getTokenAccountPools
       parameters:
       - description: The key for the token account. The exact format may vary based
-          on the token connector use.
+          on the token connector use
         in: path
         name: key
         required: true
diff --git a/docs/tutorials/custom_contracts/ethereum.md b/docs/tutorials/custom_contracts/ethereum.md
index 28a609b47..77ff405d1 100644
--- a/docs/tutorials/custom_contracts/ethereum.md
+++ b/docs/tutorials/custom_contracts/ethereum.md
@@ -549,7 +549,7 @@ Here is an example of sending 100 wei with a transaction:
 
 Now that we've seen how to submit transactions and preform read-only queries to the blockchain, let's look at how to receive blockchain events so we know when things are happening in realtime.
 
-If you look at the source code for the smart contract we're working with above, you'll notice that it emits an event when the stored value of the integer is set. In order to receive these events, we first need to instruct FireFly to listen for this specific type of blockchain event. To do this, we create an **Event Listener**. The `/contracts/listeners` endpoint is RESTful so there are `POST`, `GET`, and `DELETE` methods available on it. To create a new listener, we will make a `POST` request. We are going to tell FireFly to listen to events with name `"Changed"` from the FireFly Interface we defined earlier, referenced by its ID. We will also tell FireFly which contract address we expect to emit these events.
+If you look at the source code for the smart contract we're working with above, you'll notice that it emits an event when the stored value of the integer is set. In order to receive these events, we first need to instruct FireFly to listen for this specific type of blockchain event. To do this, we create an **Event Listener**. The `/contracts/listeners` endpoint is RESTful so there are `POST`, `GET`, and `DELETE` methods available on it. To create a new listener, we will make a `POST` request. We are going to tell FireFly to listen to events with name `"Changed"` from the FireFly Interface we defined earlier, referenced by its ID. We will also tell FireFly which contract address we expect to emit these events, and the topic to assign these events to. Topics are a way for applications to subscribe to events they are interested in.
 
 ### Request
 
@@ -565,8 +565,9 @@ If you look at the source code for the smart contract we're working with above,
   },
   "eventPath": "Changed",
   "options": {
-    "firstEvent": "oldest"
-  }
+    "firstEvent": "new"
+  },
+  "topic": "simple-storage"
 }
 ```
 
@@ -620,6 +621,69 @@ If you look at the source code for the smart contract we're working with above,
 
 We can see in the response, that FireFly pulls all the schema information from the FireFly Interface that we broadcasted earlier and creates the listener with that schema. This is useful so that we don't have to enter all of that data again.
 
+
+### Querying listener status
+
+If you are interested in learning about the current state of a listener you have created, you can query with the `fetchstatus` parameter. For FireFly stacks with an EVM compatible blockchain connector, the response will include checkpoint information and if the listener is currently in catchup mode.
+
+#### Request / Response
+
+`GET` `http://localhost:5000/api/v1/namespaces/default/contracts/listeners/1bfa3b0f-3d90-403e-94a4-af978d8c5b14?fetchstatus`
+```json
+{
+  "id": "1bfa3b0f-3d90-403e-94a4-af978d8c5b14",
+  "interface": {
+    "id": "8bdd27a5-67c1-4960-8d1e-7aa31b9084d3"
+  },
+  "namespace": "default",
+  "name": "sb-66209ffc-d355-4ac0-7151-bc82490ca9df",
+  "protocolId": "sb-66209ffc-d355-4ac0-7151-bc82490ca9df",
+  "location": {
+    "address": "0xa5ea5d0a6b2eaf194716f0cc73981939dca26da1"
+  },
+  "created": "2022-02-17T22:02:36.34549538Z",
+  "event": {
+    "name": "Changed",
+    "description": "",
+    "params": [
+      {
+        "name": "from",
+        "schema": {
+          "type": "string",
+          "details": {
+            "type": "address",
+            "internalType": "address",
+            "indexed": true
+          }
+        }
+      },
+      {
+        "name": "value",
+        "schema": {
+          "type": "integer",
+          "details": {
+            "type": "uint256",
+            "internalType": "uint256"
+          }
+        }
+      }
+    ]
+  },
+  "status": {
+    "checkpoint": {
+        "block": 0,
+        "transactionIndex": -1,
+        "logIndex": -1
+    },
+    "catchup": true
+  },
+  "options": {
+    "firstEvent": "oldest"
+  }
+}
+```
+
+
 ## Subscribe to events from our contract
 
 Now that we've told FireFly that it should listen for specific events on the blockchain, we can set up a **Subscription** for FireFly to send events to our app. To set up our subscription, we will make a `POST` to the `/subscriptions` endpoint.
@@ -747,6 +811,36 @@ You can see in the event received over the WebSocket connection, the blockchain
 
 In the event, we can also see the `blockchainevent` itself, which has an `output` object. These are the `params` in our FireFly Interface, and the actual output of the event. Here we can see the `value` is `3` which is what we set the integer to in our original transaction.
 
+### Subscription offset
+If you query by the ID of your subscription with the `fetchstatus` parameter, you can see its current `offset`.
+
+`GET` `http://localhost:5000/api/v1/namespaces/default/subscriptions/f826269c-65ed-4634-b24c-4f399ec53a32`
+```json
+{
+  "id": "f826269c-65ed-4634-b24c-4f399ec53a32",
+  "namespace": "default",
+  "name": "simple-storage",
+  "transport": "websockets",
+  "filter": {
+    "events": "blockchain_event_received",
+    "message": {},
+    "transaction": {},
+    "blockchainevent": {
+      "listener": "1bfa3b0f-3d90-403e-94a4-af978d8c5b14"
+    }
+  },
+  "options": {
+    "firstEvent": "-1",
+    "withData": false
+  },
+  "status": {
+    "offset": 20
+  }
+  "created": "2022-03-15T17:35:30.131698921Z",
+  "updated": null
+}
+```
+
 **You've reached the end of the main guide to working with custom smart contracts in FireFly**. Hopefully this was helpful and gives you what you need to get up and running with your own contracts. There are several additional ways to invoke or query smart contracts detailed below, so feel free to keep reading if you're curious.
 
 ## Appendix I: Work with a custom contract without creating a named API
diff --git a/docs/tutorials/events.md b/docs/tutorials/events.md
index c87293eb9..0bf3edc1a 100644
--- a/docs/tutorials/events.md
+++ b/docs/tutorials/events.md
@@ -134,7 +134,7 @@ it received):
 { "type": "ack", "id": "617db63-2cf5-4fa3-8320-46150cbb5372" }
 ```
 
-> _You must send an acknowledgement for every message, or you will stop receiving messages.
+> _You must send an acknowledgement for every message, or you will stop receiving messages_.
 
 ### Set up the WebSocket subscription
 
@@ -180,3 +180,7 @@ Example connection URL:
 - `namespace=default` - event listeners are scoped to a namespace
 - `name=app1` - the subscription name
 
+
+## Custom Contract Events
+
+If you are interested in learning more about events for custom smart contracts, please see the [Working with custom smart contracts](./custom_contracts/index.md) section.
diff --git a/internal/apiserver/route_get_contract_listener_by_name_or_id.go b/internal/apiserver/route_get_contract_listener_by_name_or_id.go
index 7424515f0..4408fe7d3 100644
--- a/internal/apiserver/route_get_contract_listener_by_name_or_id.go
+++ b/internal/apiserver/route_get_contract_listener_by_name_or_id.go
@@ -18,6 +18,7 @@ package apiserver
 
 import (
 	"net/http"
+	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -32,7 +33,9 @@ var getContractListenerByNameOrID = &ffapi.Route{
 	PathParams: []*ffapi.PathParam{
 		{Name: "nameOrId", Description: coremsgs.APIParamsContractListenerNameOrID},
 	},
-	QueryParams:     nil,
+	QueryParams: []*ffapi.QueryParam{
+		{Name: "fetchstatus", Example: "true", Description: coremsgs.APIParamsFetchStatus, IsBool: true},
+	},
 	Description:     coremsgs.APIEndpointsGetContractListenerByNameOrID,
 	JSONInputValue:  nil,
 	JSONOutputValue: func() interface{} { return &core.ContractListener{} },
@@ -42,6 +45,9 @@ var getContractListenerByNameOrID = &ffapi.Route{
 			return or.Contracts() != nil
 		},
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
+			if strings.EqualFold(r.QP["fetchstatus"], "true") {
+				return cr.or.Contracts().GetContractListenerByNameOrIDWithStatus(cr.ctx, r.PP["nameOrId"])
+			}
 			return cr.or.Contracts().GetContractListenerByNameOrID(cr.ctx, r.PP["nameOrId"])
 		},
 	},
diff --git a/internal/apiserver/route_get_subscription_by_id.go b/internal/apiserver/route_get_subscription_by_id.go
index e099ce7b4..3b29b7960 100644
--- a/internal/apiserver/route_get_subscription_by_id.go
+++ b/internal/apiserver/route_get_subscription_by_id.go
@@ -18,6 +18,7 @@ package apiserver
 
 import (
 	"net/http"
+	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -31,15 +32,19 @@ var getSubscriptionByID = &ffapi.Route{
 	PathParams: []*ffapi.PathParam{
 		{Name: "subid", Description: coremsgs.APIParamsSubscriptionID},
 	},
-	QueryParams:     nil,
+	QueryParams: []*ffapi.QueryParam{
+		{Name: "fetchstatus", Description: coremsgs.APIParamsFetchStatus, IsBool: true},
+	},
 	Description:     coremsgs.APIEndpointsGetSubscriptionByID,
 	JSONInputValue:  nil,
 	JSONOutputValue: func() interface{} { return &core.Subscription{} },
 	JSONOutputCodes: []int{http.StatusOK},
 	Extensions: &coreExtensions{
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
-			output, err = cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
-			return output, err
+			if strings.EqualFold(r.QP["fetchstatus"], "true") {
+				return cr.or.GetSubscriptionByIDWithStatus(cr.ctx, r.PP["subid"])
+			}
+			return cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
 		},
 	},
 }
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 6e2e15bf7..3c112ee48 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -728,6 +728,19 @@ func (e *Ethereum) DeleteContractListener(ctx context.Context, subscription *cor
 	return e.streams.deleteSubscription(ctx, subscription.BackendID)
 }
 
+func (e *Ethereum) GetContractListenerStatus(ctx context.Context, subID string) (status *fftypes.JSONAny, err error) {
+	sub, err := e.streams.getSubscription(ctx, subID)
+	if err != nil {
+		return nil, err
+	}
+
+	normalized, err := json.Marshal(sub.subscriptionCheckpoint)
+	if err == nil {
+		status = fftypes.JSONAnyPtrBytes(normalized)
+	}
+	return status, nil
+}
+
 func (e *Ethereum) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	return &ffi2abi.ParamValidator{}, nil
 }
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index 5979f3eee..6010100bd 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/ffresty"
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-signer/pkg/abi"
@@ -51,12 +52,19 @@ type eventStream struct {
 }
 
 type subscription struct {
-	ID        string     `json:"id"`
-	Name      string     `json:"name,omitempty"`
-	Stream    string     `json:"stream"`
-	FromBlock string     `json:"fromBlock"`
-	Address   string     `json:"address"`
-	Event     *abi.Entry `json:"event"`
+	ID               string            `json:"id"`
+	Name             string            `json:"name,omitempty"`
+	Stream           string            `json:"stream"`
+	FromBlock        string            `json:"fromBlock"`
+	EthCompatAddress string            `json:"address,omitempty"`
+	EthCompatEvent   *abi.Entry        `json:"event,omitempty"`
+	Filters          []fftypes.JSONAny `json:"filters"`
+	subscriptionCheckpoint
+}
+
+type subscriptionCheckpoint struct {
+	Checkpoint *fftypes.JSONAny `json:"checkpoint,omitempty"`
+	Catchup    bool             `json:"catchup,omitempty"`
 }
 
 func newStreamManager(client *resty.Client, cache *ccache.Cache, cacheTTL time.Duration) *streamManager {
@@ -181,11 +189,11 @@ func (s *streamManager) createSubscription(ctx context.Context, location *Locati
 		fromBlock = "latest"
 	}
 	sub := subscription{
-		Name:      subName,
-		Stream:    stream,
-		FromBlock: fromBlock,
-		Address:   location.Address,
-		Event:     abi,
+		Name:             subName,
+		Stream:           stream,
+		FromBlock:        fromBlock,
+		EthCompatAddress: location.Address,
+		EthCompatEvent:   abi,
 	}
 	res, err := s.client.R().
 		SetContext(ctx).
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index bd3d54442..8ac782ca0 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -833,6 +833,11 @@ func (f *Fabric) DeleteContractListener(ctx context.Context, subscription *core.
 	return f.streams.deleteSubscription(ctx, subscription.BackendID)
 }
 
+func (f *Fabric) GetContractListenerStatus(ctx context.Context, subID string) (*fftypes.JSONAny, error) {
+	// Fabconnect does not currently provide any additional status info for listener subscriptions
+	return nil, nil
+}
+
 func (f *Fabric) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	// Fabconnect does not require any additional validation beyond "JSON Schema correctness" at this time
 	return nil, nil
diff --git a/internal/contracts/manager.go b/internal/contracts/manager.go
index e0fe6ddd6..01559f4e1 100644
--- a/internal/contracts/manager.go
+++ b/internal/contracts/manager.go
@@ -54,6 +54,7 @@ type Manager interface {
 	AddContractListener(ctx context.Context, listener *core.ContractListenerInput) (output *core.ContractListener, err error)
 	AddContractAPIListener(ctx context.Context, apiName, eventPath string, listener *core.ContractListener) (output *core.ContractListener, err error)
 	GetContractListenerByNameOrID(ctx context.Context, nameOrID string) (*core.ContractListener, error)
+	GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (*core.ContractListenerWithStatus, error)
 	GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error)
 	GetContractAPIListeners(ctx context.Context, apiName, eventPath string, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error)
 	DeleteContractListenerByNameOrID(ctx context.Context, nameOrID string) error
@@ -572,6 +573,22 @@ func (cm *contractManager) GetContractListenerByNameOrID(ctx context.Context, na
 	return listener, nil
 }
 
+func (cm *contractManager) GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (enrichedListener *core.ContractListenerWithStatus, err error) {
+	listener, err := cm.GetContractListenerByNameOrID(ctx, nameOrID)
+	if err != nil {
+		return nil, err
+	}
+	status, err := cm.blockchain.GetContractListenerStatus(ctx, listener.BackendID)
+	if err != nil {
+		return nil, err
+	}
+	enrichedListener = &core.ContractListenerWithStatus{
+		ContractListener: *listener,
+		Status:           status,
+	}
+	return enrichedListener, nil
+}
+
 func (cm *contractManager) GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error) {
 	return cm.database.GetContractListeners(ctx, cm.namespace, filter)
 }
diff --git a/internal/coremsgs/en_api_translations.go b/internal/coremsgs/en_api_translations.go
index 7a81972b7..5548926a9 100644
--- a/internal/coremsgs/en_api_translations.go
+++ b/internal/coremsgs/en_api_translations.go
@@ -50,7 +50,7 @@ var (
 	APIParamsDID                            = ffm("api.params.DID", "The identity DID")
 	APIParamsNodeNameOrID                   = ffm("api.params.nodeNameOrID", "The name or ID of the node")
 	APIParamsOrgNameOrID                    = ffm("api.params.orgNameOrID", "The name or ID of the org")
-	APIParamsTokenAccountKey                = ffm("api.params.tokenAccountKey", "The key for the token account. The exact format may vary based on the token connector use.")
+	APIParamsTokenAccountKey                = ffm("api.params.tokenAccountKey", "The key for the token account. The exact format may vary based on the token connector use")
 	APIParamsTokenPoolNameOrID              = ffm("api.params.tokenPoolNameOrID", "The token pool name or ID")
 	APIParamsTokenTransferFromOrTo          = ffm("api.params.tokenTransferFromOrTo", "The sending or receiving token account for a token transfer")
 	APIParamsTokenTransferID                = ffm("api.params.tokenTransferID", "The token transfer ID")
@@ -63,6 +63,7 @@ var (
 	APIParamsMetadata                       = ffm("api.params.metadata", "Metadata associated with this data item")
 	APIParamsAutometa                       = ffm("api.params.autometa", "When set, FireFly will automatically generate JSON metadata with the upload details")
 	APIParamsContractAPIID                  = ffm("api.params.contractAPIID", "The ID of the contract API")
+	APIParamsFetchStatus                    = ffm("api.params.fetchStatus", "When set, the API will return additional status information if available")
 
 	APIEndpointsAdminGetNamespaceByName = ffm("api.endpoints.adminGetNamespaceByName", "Gets a namespace by name")
 	APIEndpointsAdminGetNamespaces      = ffm("api.endpoints.adminGetNamespaces", "List namespaces")
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index 1f9188c93..912d245cb 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -262,4 +262,5 @@ var (
 	MsgInvalidConnectorName               = ffe("FF10420", "Could not find name %s for %s connector")
 	MsgCannotInitLegacyNS                 = ffe("FF10421", "could not initialize legacy '%s' namespace - found conflicting V1 multi-party config in %s and %s")
 	MsgInvalidGroupMember                 = ffe("FF10422", "invalid group member - node '%s' is not owned by '%s' or any of its ancestors")
+	MsgContractListenerStatusInvalid      = ffe("FF10423", "Failed to validate contract listener status: %v", 400)
 )
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 9e537add8..92f4533eb 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -74,6 +74,7 @@ type Orchestrator interface {
 	// Subscription management
 	GetSubscriptions(ctx context.Context, filter database.AndFilter) ([]*core.Subscription, *database.FilterResult, error)
 	GetSubscriptionByID(ctx context.Context, id string) (*core.Subscription, error)
+	GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error)
 	CreateSubscription(ctx context.Context, subDef *core.Subscription) (*core.Subscription, error)
 	CreateUpdateSubscription(ctx context.Context, subDef *core.Subscription) (*core.Subscription, error)
 	DeleteSubscription(ctx context.Context, id string) error
diff --git a/internal/orchestrator/subscriptions.go b/internal/orchestrator/subscriptions.go
index 742de7a80..af12327d3 100644
--- a/internal/orchestrator/subscriptions.go
+++ b/internal/orchestrator/subscriptions.go
@@ -76,3 +76,30 @@ func (or *orchestrator) GetSubscriptionByID(ctx context.Context, id string) (*co
 	}
 	return or.database().GetSubscriptionByID(ctx, or.namespace.Name, u)
 }
+
+func (or *orchestrator) GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error) {
+	sub, err := or.GetSubscriptionByID(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	if sub == nil {
+		return nil, nil
+	}
+
+	offset, err := or.database().GetOffset(ctx, core.OffsetTypeSubscription, sub.ID.String())
+	if err != nil {
+		return nil, err
+	}
+
+	subWithStatus := &core.SubscriptionWithStatus{
+		Subscription: *sub,
+	}
+
+	if offset != nil {
+		subWithStatus.Status = core.SubscriptionStatus{
+			CurrentOffset: offset.Current,
+		}
+	}
+
+	return subWithStatus, nil
+}
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index 50dfe3797..920a563f2 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -154,6 +154,29 @@ func (_m *Plugin) GetAndConvertDeprecatedContractConfig(ctx context.Context) (*f
 	return r0, r1, r2
 }
 
+// GetContractListenerStatus provides a mock function with given fields: ctx, subID
+func (_m *Plugin) GetContractListenerStatus(ctx context.Context, subID string) (*fftypes.JSONAny, error) {
+	ret := _m.Called(ctx, subID)
+
+	var r0 *fftypes.JSONAny
+	if rf, ok := ret.Get(0).(func(context.Context, string) *fftypes.JSONAny); ok {
+		r0 = rf(ctx, subID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*fftypes.JSONAny)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, subID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetFFIParamValidator provides a mock function with given fields: ctx
 func (_m *Plugin) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	ret := _m.Called(ctx)
diff --git a/mocks/contractmocks/manager.go b/mocks/contractmocks/manager.go
index 1924f395f..0b0125bbb 100644
--- a/mocks/contractmocks/manager.go
+++ b/mocks/contractmocks/manager.go
@@ -235,6 +235,29 @@ func (_m *Manager) GetContractListenerByNameOrID(ctx context.Context, nameOrID s
 	return r0, r1
 }
 
+// GetContractListenerByNameOrIDWithStatus provides a mock function with given fields: ctx, nameOrID
+func (_m *Manager) GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (*core.ContractListenerWithStatus, error) {
+	ret := _m.Called(ctx, nameOrID)
+
+	var r0 *core.ContractListenerWithStatus
+	if rf, ok := ret.Get(0).(func(context.Context, string) *core.ContractListenerWithStatus); ok {
+		r0 = rf(ctx, nameOrID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.ContractListenerWithStatus)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, nameOrID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetContractListeners provides a mock function with given fields: ctx, filter
 func (_m *Manager) GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error) {
 	ret := _m.Called(ctx, filter)
diff --git a/mocks/orchestratormocks/orchestrator.go b/mocks/orchestratormocks/orchestrator.go
index e2e092127..d3ae3ef3a 100644
--- a/mocks/orchestratormocks/orchestrator.go
+++ b/mocks/orchestratormocks/orchestrator.go
@@ -971,6 +971,29 @@ func (_m *Orchestrator) GetSubscriptionByID(ctx context.Context, id string) (*co
 	return r0, r1
 }
 
+// GetSubscriptionByIDWithStatus provides a mock function with given fields: ctx, id
+func (_m *Orchestrator) GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *core.SubscriptionWithStatus
+	if rf, ok := ret.Get(0).(func(context.Context, string) *core.SubscriptionWithStatus); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.SubscriptionWithStatus)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetSubscriptions provides a mock function with given fields: ctx, filter
 func (_m *Orchestrator) GetSubscriptions(ctx context.Context, filter database.AndFilter) ([]*core.Subscription, *database.FilterResult, error) {
 	ret := _m.Called(ctx, filter)
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index 0af67bea0..1c490a03d 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -75,6 +75,8 @@ type Plugin interface {
 	// DeleteContractListener deletes a previously-created subscription
 	DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error
 
+	GetContractListenerStatus(ctx context.Context, subID string) (*fftypes.JSONAny, error)
+
 	// GetFFIParamValidator returns a blockchain-plugin-specific validator for FFIParams and their JSON Schema
 	GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error)
 
diff --git a/pkg/core/contract_listener.go b/pkg/core/contract_listener.go
index d68e65b25..0c143e9e1 100644
--- a/pkg/core/contract_listener.go
+++ b/pkg/core/contract_listener.go
@@ -39,6 +39,10 @@ type ContractListener struct {
 	Options   *ContractListenerOptions `ffstruct:"ContractListener" json:"options,omitempty"`
 }
 
+type ContractListenerWithStatus struct {
+	ContractListener
+	Status *fftypes.JSONAny `ffstruct:"ContractListenerWithStatus" json:"status,omitempty" ffexcludeinput:"true"`
+}
 type ContractListenerOptions struct {
 	FirstEvent string `ffstruct:"ContractListenerOptions" json:"firstEvent,omitempty"`
 }
diff --git a/pkg/core/subscription.go b/pkg/core/subscription.go
index 95acb1278..97e80e27f 100644
--- a/pkg/core/subscription.go
+++ b/pkg/core/subscription.go
@@ -122,6 +122,15 @@ type Subscription struct {
 	Updated   *fftypes.FFTime     `ffstruct:"Subscription" json:"updated" ffexcludeinput:"true"`
 }
 
+type SubscriptionWithStatus struct {
+	Subscription
+	Status SubscriptionStatus `ffstruct:"SubscriptionWithStatus" json:"status,omitempty" ffexcludeinput:"true"`
+}
+
+type SubscriptionStatus struct {
+	CurrentOffset int64 `ffstruct:"SubscriptionStatus" json:"currentOffset,omitempty" ffexcludeinout:"true"`
+}
+
 func (so *SubscriptionOptions) UnmarshalJSON(b []byte) error {
 	so.additionalOptions = fftypes.JSONObject{}
 	err := json.Unmarshal(b, &so.additionalOptions)


Test patch:
diff --git a/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go b/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
index f7ba336db..b5ec429de 100644
--- a/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
+++ b/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
@@ -43,3 +43,20 @@ func TestGetContractListenerByNameOrID(t *testing.T) {
 
 	assert.Equal(t, 200, res.Result().StatusCode)
 }
+
+func TestGetContractListenerByNameOrIDWithStatus(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	mcm := &contractmocks.Manager{}
+	o.On("Contracts").Return(mcm)
+	id := fftypes.NewUUID()
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/contracts/listeners/"+id.String()+"?fetchstatus", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+
+	mcm.On("GetContractListenerByNameOrIDWithStatus", mock.Anything, id.String()).
+		Return(&core.ContractListenerWithStatus{}, nil)
+	r.ServeHTTP(res, req)
+
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
diff --git a/internal/apiserver/route_get_subscription_by_id_test.go b/internal/apiserver/route_get_subscription_by_id_test.go
index d1a8d3337..6c34a92a7 100644
--- a/internal/apiserver/route_get_subscription_by_id_test.go
+++ b/internal/apiserver/route_get_subscription_by_id_test.go
@@ -38,3 +38,19 @@ func TestGetSubscriptionByID(t *testing.T) {
 
 	assert.Equal(t, 200, res.Result().StatusCode)
 }
+
+func TestGetSubscriptionByIDWithStatus(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?fetchstatus", 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("GetSubscriptionByIDWithStatus", mock.Anything, "abcd12345").
+		Return(&core.SubscriptionWithStatus{}, nil)
+	r.ServeHTTP(res, req)
+
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index bdb80e10e..b6a1b9020 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -2997,7 +2997,6 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 		}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3039,7 +3038,6 @@ func TestAddFireflySubscriptionV1(t *testing.T) {
 		}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3076,7 +3074,6 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3113,7 +3110,6 @@ func TestAddFireflySubscriptionCreateError(t *testing.T) {
 	httpmock.RegisterResponder("POST", "http://localhost:12345/",
 		httpmock.NewJsonResponderOrPanic(500, `pop`))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3149,7 +3145,6 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(500, `pop`))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3165,3 +3160,74 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10111", err)
 }
+
+func TestGetContractListenerStatus(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	resetConf(e)
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	checkpoint := fftypes.JSONAnyPtr(fftypes.JSONObject{
+		"block":            0,
+		"transactionIndex": -1,
+		"logIndex":         -1,
+	}.String())
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions/sub1",
+		httpmock.NewJsonResponderOrPanic(200, subscription{
+			ID: "sub1", Stream: "es12345", Name: "ff-sub-1132312312312", subscriptionCheckpoint: subscriptionCheckpoint{
+				Catchup:    false,
+				Checkpoint: checkpoint,
+			},
+		}))
+
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics)
+	assert.NoError(t, err)
+
+	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	assert.NotNil(t, status)
+	assert.NoError(t, err)
+}
+
+func TestGetContractListenerStatusGetSubFail(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	resetConf(e)
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions/sub1",
+		httpmock.NewJsonResponderOrPanic(500, `pop`))
+
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics)
+	assert.NoError(t, err)
+
+	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	assert.Nil(t, status)
+	assert.Regexp(t, "FF10111", err)
+}
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index d9cf65dee..453cd452e 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -2519,3 +2519,14 @@ func TestSubmitNetworkActionVersionError(t *testing.T) {
 	err := e.SubmitNetworkAction(context.Background(), "", signer, core.NetworkActionTerminate, location)
 	assert.Regexp(t, "FF10284", err)
 }
+
+func TestGetContractListenerStatus(t *testing.T) {
+	e, cancel := newTestFabric()
+	defer cancel()
+	httpmock.ActivateNonDefault(e.client.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	status, err := e.GetContractListenerStatus(context.Background(), "id")
+	assert.Nil(t, status)
+	assert.NoError(t, err)
+}
diff --git a/internal/contracts/manager_test.go b/internal/contracts/manager_test.go
index 81ecfd11f..34ac51736 100644
--- a/internal/contracts/manager_test.go
+++ b/internal/contracts/manager_test.go
@@ -1579,6 +1579,45 @@ func TestGetContractListenerByNameOrID(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestGetContractListenerByNameOrIDWithStatus(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+
+	id := fftypes.NewUUID()
+	backendID := "testID"
+	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(&core.ContractListener{BackendID: backendID}, nil)
+	mbi.On("GetContractListenerStatus", context.Background(), backendID).Return(fftypes.JSONAnyPtr(fftypes.JSONObject{}.String()), nil)
+
+	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+	assert.NoError(t, err)
+}
+
+func TestGetContractListenerByNameOrIDWithStatusListenerFail(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+
+	id := fftypes.NewUUID()
+	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(nil, fmt.Errorf("pop"))
+
+	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+	assert.EqualError(t, err, "pop")
+}
+
+func TestGetContractListenerByNameOrIDWithStatusPluginFail(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+
+	id := fftypes.NewUUID()
+	backendID := "testID"
+	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(&core.ContractListener{BackendID: backendID}, nil)
+	mbi.On("GetContractListenerStatus", context.Background(), backendID).Return(nil, fmt.Errorf("pop"))
+
+	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+	assert.EqualError(t, err, "pop")
+}
+
 func TestGetContractListenerByNameOrIDFail(t *testing.T) {
 	cm := newTestContractManager()
 	mdi := cm.database.(*databasemocks.Plugin)
diff --git a/internal/orchestrator/subscriptions_test.go b/internal/orchestrator/subscriptions_test.go
index fd6afb422..f10e41829 100644
--- a/internal/orchestrator/subscriptions_test.go
+++ b/internal/orchestrator/subscriptions_test.go
@@ -157,3 +157,63 @@ func TestGetSubscriptionDefsByIDBadID(t *testing.T) {
 	_, err := or.GetSubscriptionByID(context.Background(), "")
 	assert.Regexp(t, "FF00138", err)
 }
+
+func TestGetSGetSubscriptionsByIDWithStatus(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	sub := &core.Subscription{
+		SubscriptionRef: core.SubscriptionRef{
+			ID:        u,
+			Name:      "sub1",
+			Namespace: "ns1",
+		},
+	}
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(sub, nil)
+	or.mdi.On("GetOffset", context.Background(), core.OffsetTypeSubscription, u.String()).Return(&core.Offset{Current: 100}, nil)
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.NoError(t, err)
+	assert.NotNil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusQuerySubFail(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(nil, fmt.Errorf("pop"))
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.EqualError(t, err, "pop")
+	assert.Nil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusOffsetQueryError(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	sub := &core.Subscription{
+		SubscriptionRef: core.SubscriptionRef{
+			ID:        u,
+			Name:      "sub1",
+			Namespace: "ns1",
+		},
+	}
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(sub, nil)
+	or.mdi.On("GetOffset", context.Background(), core.OffsetTypeSubscription, u.String()).Return(nil, fmt.Errorf("pop"))
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.EqualError(t, err, "pop")
+	assert.Nil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusUnknownSub(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(nil, nil)
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.NoError(t, err)
+	assert.Nil(t, subWithStatus)
+}