Subscriptions and Filtering

[darnjo]

Summary

At the Summer 2023 Developers Workshop there was some discussion about extending the EntityEvent Resource (RCP-027) to support a) event types and b) payloads. The addition of event types were approved by the group at the Dev Workshop but the addition of payloads was not.

For some background, much care was taken in the initial RCP-027 proposal to avoid writing data to the log that might expose sensitive information about a particular event. As such, the original proposal intentionally avoided both event types and payloads.

However, RCP-027 is meant to be a framework for any event stream, and providers can add this data as they see fit. The recommendation would be that they do so in a way that matches existing Data Dictionary JSON Schema (i.e. RESO Common Format) rather than creating a new non-standard payload shape.

EntityEvent Resource and Filtering

The EntityEvent Resource was initially created to support replication, in which case the consumer would pull all events they had access to and fetch their corresponding records. As such, event types weren't added to the original proposal.

However, there are some cases where the user only wants a subset of the events available and wants to filter based on their event types.

New Addition to RCP-027: EventTypes Array

The group at the workshop requested that an EventTypes array be added to the current EntityEvent Resource, which would be "Open with Enumerations." RESO will provide a few well-known lookups that don't reveal information to the user (e.g. StatusChanged rather than ClosedListing), and providers may extend EventTypes with their own enumerations if they want to provide more information than that.

Example - Filter by Canceled Open Houses

REQUEST

GET /EntityEvent?$filter=EntityEventSequence gt 100 and ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')

RESPONSE

{
  "@odata.context": "/EntityEvent?$filter=EntityEventSequence gt 100 and ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')",
  "value": [
    {
      "EntityEventSequence": 101,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": "21"
    },
    {
      "EntityEventSequence": 103,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": "539"
    },
    {
      "EntityEventSequence": 110,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": "1239"
    }
  ]
}

Since EventTypes is an array, clients can filter on multiple event types at once.

Webhooks, Filtering, and Subscriptions

There is also interest in supporting webhooks with the EntityEvent Resource, and there's an initial proposal for this in RCP-028 that was approved in 2020. Further extension is needed in order to support subscriptions, filtering, and multiple event streams.

EntityEventSubscription Resource

Consumers need a way to differentiate between multiple webhooks feeds when they're receiving events. For example, assume there are two feeds - "IDX" and "OpenHouse Canceled" - going to the same destination. A consumer receiving these events needs to be able to tell them apart.

A "Consumer Label" was originally added to the headers in RCP-028 to accomplish this goal. The reason it was added to the headers rather than the payload initially was that EntityEvent does not have a concept of subscriptions or feed types, and it could change on a per-subscription basis. There was some discussion at the Dev Workshop about how this should be handled and more research is still needed.

One thing that RCP-028 doesn't do is establish a way to create new EntityEvent push subscriptions. These may have a filter, or could be for the entire feed that a consumer has access to (or both).

At the Dev Workshop, what was discussed was adding a new resource to manage subscriptions, called EntityEventSubscription.

Example - Create New Subscription with Filter

REQUEST

POST /EntityEventSubscription
{
  "EntityEventFilter": "ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')",
  "EntityEventSequence": 103,
  "CallbackUrl": "https://my.api.com/feeds",
  "CallbackCredentialId": "credentials1234" 
}

Where each item is defined as follows:

• EntityEventFilter - The OData-compliant filter used on the EntityEvent resource, without the sequence number.
• EntityEventSequence - The last sequence number that the client received (if omitted, the subscription would start at the current event).
• CallbackUrl - The URL for the EntityEvent Subscription to post data to when events are found.
• CallbackCredentialsId - This wasn't discussed at the Dev Workshop, but the receiving server is unlikely to have an open endpoint to push events to, so some kind of auth is needed. Would that be provided in the subscription or somewhere else? In order to keep things more secure, perhaps credentials are added in another process and their id is referenced in the subscription process instead of the raw credentials? Or is it OK to store them along with the subscription?

RESPONSE

HTTP/2 201 Created
  OData-Version: 4.01
  EntityId: "12345"
  Location: https://api.reso.org/EntityEventSubscription('12345')
  Content-Length: 243
  Content-Type: application/json
  Preference-Applied: return=representation

{
  "@odata.context": "https://api.reso.org/$metadata#EntityEventSubscription/$entity",
  "@odata.id": "https://api.reso.org/EntityEventSubscription('12345')",
  "@odata.editLink": "https://api.reso.org/EntityEventSubscription('12345')",
  "@odata.etag": "W/\"MjAxOC0wMS0yM1QwODo1Njo0NS4yMi0wODowMA==\"",
  "EntityEventSubscriptionKey": "12345",
  "EntityEventFilter": "ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')",
  "EntityEventSequence": 103,
  "CallbackUrl": "https://my.api.com/feeds",
  "CallbackCredentialId": "credentials1234",
  "ModificationTimestamp": "2022-12-05T18:33:20Z"
}

Example - Resume Feed After Interruption

With webhooks and event subscriptions, it's possible that the consumer API isn't available at the time that the producer tries and posts new events. It's also possible that the producer could supply events faster than the consumer can process them (slow consumer). Some of this is outlined in the "Polite Behavior" section in the original RCP-028 specification.

Producers may use retries and dead letter queues to help provide more robust handling for the consumer in these cases, which probably needs further discussion as to whether these attributes are communicated in the subscription. But at some point the producer can only do so much to support the consumer, the feed may be stopped, and may need to be reinitialized by the consumer.

The assumption with both the RCP-027 and RCP-028 proposals is that it's the consumer's responsibility to remember where they were in the feed so they can resume.

The producer could also take on this responsibility if they're tracking (2xx) status codes for each event sent, but this means additional work and state information that the producer needs to track, which could add up with large event subscriptions and many feeds and consumers.

This probably needs further discussion as well, and for now we'll assume that it's the consumer's responsibility. Not much changes if the producer decides to take on this role instead - the last sequence number wouldn't be needed as the producer could supply it.

The consumer might still want to reinitialize at some point, for whatever reason, which means making the following request.

REQUEST

PATCH https://api.reso.org/EntityEventSubscription('12345')
  OData-Version: 4.01
  Content-Type: application/json
  Accept: application/json
  If-Match: W/\"MjAxOC0wMS0yM1QwODo1Njo0NS4yMi0wODowMA==\"
  Prefer: return=representation

{
  "EntityEventSequence": 100
}

RESPONSE

{
  "@odata.context": "https://api.reso.org/$metadata#EntityEventSubscription/$entity",
  "@odata.id": "https://api.reso.org/EntityEventSubscription('12345')",
  "@odata.editLink": "https://api.reso.org/EntityEventSubscription('12345')",
  "@odata.etag": "W/\"MjAxOC0wMS0yM1QwODo1Njo0NS4yMi0wODowMA==\"",
  "EntityEventSubscriptionKey": "12345",
  "EntityEventFilter": "ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')",
  "EntityEventSequence": 100,
  "CallbackUrl": "https://my.api.com/feeds",
  "CallbackCredentialId": "credentials1234",
  "ModificationTimestamp": "2023-07-05T18:33:20Z"
}

Note that the EntityEventSequence might not be present for a given request. For example, if the provider was using compaction or they no longer had access to the record. The consumer would therefore post the event they want, and the server would return the first event greater than or equal to 100 that the consumer had access to.

Removing Subscriptions

In order to remove the subscription, a client would issue the following request:

REQUEST

DELETE https://api.reso.org/EntityEventSubscription('12345')
  OData-Version: 4.01
  If-Match: W/\"MjAxOC0wMS0yM1QwODo1Njo0NS4yMi0wODowMA==\"

RESPONSE

HTTP/2 204 No Content
 OData-Version: 4.01

Deactivating Subscriptions

There may be a need to stop an EntityEventSubscription or indicate it's in/active. If so, an additional field, such as ActiveYN, might be appropriate.

Updating this field would follow the Update semantics outlined in RCP-010, and above.

[SergioDelRioT4Bi]

I think there is an error in the response output:

{
  "@odata.context": "/EntityEvent?$filter=EntityEventSequence gt 100 and ResourceName eq 'OpenHouse' and EventTypes/any(EventType: EventType eq 'Canceled')",
  "value": [
    {
      "EntityEventSequence": 101,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": 21
    },
    {
      "EntityEventSequence": 103,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": 539
    },
    {
      "EntityEventSequence": 110,
      "ResourceName": "OpenHouse",
      "EventTypes": ["Canceled"],
      "ResourceRecordKey": 1239
    }
  ]
}

ResouurceRecordKey should be a String should it not?