@platformatic/graphql-subscriptions-resume

An addon to @fastify/http-proxy to resume GraphQL subscriptions. This library helps manage subscriptions state across client connections, allowing clients to resume subscriptions from where they left off after reconnecting.

Installation

npm install @platformatic/graphql-subscriptions-resume

or

pnpm add @platformatic/graphql-subscriptions-resume

Key Features

• State Persistence: Track subscription state by client ID
• Resumable Subscriptions: Automatically resume subscriptions from the last received value
• GraphQL Parsing: Smart parsing of GraphQL subscription queries
• Alias Support: Handle subscription aliases properly
• Easy Integration: Works well with Platformatic proxy services or any WebSocket-based GraphQL implementation

Usage

Basic Setup

import { StatefulSubscriptions } from '@platformatic/graphql-subscriptions-resume'
import { logger } from './your-logger.js'

// Initialize with your subscription configurations
const state = new StatefulSubscriptions({
  subscriptions: [
    {
      name: 'onItems',  // The subscription name in your GraphQL schema
      key: 'offset'     // The field that represents a sequence or position (e.g., offset, timestamp, id)
    },
    {
      name: 'onNotifications',
      key: 'id'
    }
  ],
  logger: logger // Provide a pino compatible logger
})

Integration with WebSocket Handlers

// Handle client connections
function onConnect(clientId) {
  console.log(`Client ${clientId} connected`)
}

// Handle client disconnections
function onDisconnect(clientId) {
  console.log(`Client ${clientId} disconnected`)
  // Clean up subscriptions when a client disconnects
  state.removeAllSubscriptions(clientId)
}

// Handle client reconnections
function onReconnect(clientId, webSocketConnection) {
  console.log(`Client ${clientId} reconnected`)
  // Restore subscriptions to their previous state
  state.restoreSubscriptions(clientId, webSocketConnection)
}

// Process incoming subscription requests
function onIncomingMessage(clientId, message) {
  // Parse the message and handle subscription requests
  const parsedMessage = JSON.parse(message)
  
  if (parsedMessage.type === 'start') {
    try {
      // Register the subscription with the state manager
      state.addSubscription(
        clientId,
        parsedMessage.payload.query,
        parsedMessage.payload.variables
        parsedMessage.id,
        parsedMessage.type,
      )
    } catch (err) {
      console.error('Error adding subscription', err)
    }
  }
}

// Process outgoing subscription updates
function onOutgoingMessage(clientId, message) {
  const parsedMessage = JSON.parse(message)
  
  if (parsedMessage.type === 'data') {
    // Update the subscription state with the latest data
    state.updateSubscriptionState(clientId, parsedMessage.payload.data)
  }
}

Complete Example with Platformatic GraphQL Composer

Here's a complete example using the library with Platformatic's GraphQL Composer:

'use strict'

const { StatefulSubscriptions } = require('@platformatic/graphql-subscriptions-resume')

const state = new StatefulSubscriptions({
  subscriptions: [
    {
      name: 'onItems',
      key: 'offset'
    },
    {
      name: 'onNotifications',
      key: 'id',
    }
  ],
  logger: globalThis.platformatic.logger
})

const hooks = {
  onConnect: (context, source, target) => {
    context.log.debug({ clientId: source.clientId }, 'onConnect')
  },
  onDisconnect: (context, source, target) => {
    context.log.debug({ clientId: source.clientId }, 'onDisconnect (client disconnected)')
    state.removeAllSubscriptions(source.clientId)
  },
  onReconnect: (context, source, target) => {
    context.log.debug({ clientId: source.clientId }, 'onReconnect')
    state.restoreSubscriptions(source.clientId, target)
  },
  onIncomingMessage: (context, source, target, message) => {
    const m = JSON.parse(message.data.toString('utf-8'))
    context.log.debug({ m, binary: message.binary, clientId: source.clientId }, 'onIncomingMessage')
    if(!source.clientId){
      source.clientId = randomUUID()
    }

    if (!m || !m.type) {
      return
    }

    if (m.type === 'start' || m.type === 'subscribe') {
      try {
        state.addSubscription(source.clientId, m.payload.query, m.payload.variables, m.id, m.type)
      } catch (err) {
        context.log.error({ err, m, clientId: source.clientId }, 'Error adding subscription')
      }
      return
    }

    if(m.type === 'connection_init') {
      try {
        state.addSubscriptionInit(source.clientId, m.payload)
      } catch (err) {
        context.log.error({ err, payload, clientId: source.clientId }, 'Error adding subscription init')
      }
      return
    }    

    if (m.type === 'complete' || m.type === 'stop') {
      try {
        if (m.id) {
          state.removeSubscription(source.clientId, m.id)
        } else {
          state.removeAllSubscriptions(source.clientId)
        }
      } catch (err) {
        context.log.error({ err, m, clientId: source.clientId }, 'Error removing subscription')
      }
      return
    }
  },
  onOutgoingMessage: (context, source, target, message) => {
    const m = JSON.parse(message.data.toString('utf-8'))
    context.log.debug({ m, binary: message.binary, clientId: source.clientId }, 'onOutgoingMessage')

    if (m.type === 'data' || m.type === 'next') {
      try {
        state.updateSubscriptionState(source.clientId, m.payload.data)
      } catch (err) {
        context.log.error({ err, m, clientId: source.clientId }, 'Error updating subscription state')
      }
      return
    }
  }
}

module.exports = hooks

How It Works

The library works by:

1. Tracking Subscriptions: When a client initiates a subscription, the library parses the GraphQL query to extract the subscription name, fields, parameters, and aliases.

2. Monitoring Updates: As subscription data is sent to clients, the library tracks the value of the configured "key" field for each subscription.

3. Resuming After Reconnection: When a client reconnects, the library automatically generates and sends a new subscription query that includes the last received key value, allowing the subscription to resume from where it left off.

API Reference

StatefulSubscriptions

The main class for managing subscriptions.

Constructor

new StatefulSubscriptions(options: StatefulSubscriptionsOptions)

Options:

• subscriptions: Array of subscription configurations
  • name: The name of the subscription as defined in your GraphQL schema
  • key: The field name used to track the subscription's position/state
  • args: (Optional) Fixed arguments that will be included in every recovery query
• logger: A pino compatibile logger instance

Methods

addSubscription

addSubscription(clientId: string, query: string, variables?: Record<string, any>, subscriptionId?: string, subscriptionType? string): void

Registers a new subscription for a client. Parses the GraphQL query and stores information about the subscription.

Parameters:

• clientId: A unique identifier for the client
• query: The GraphQL subscription query
• variables: Optional GraphQL variables for the query
• subscriptionId: A unique identifier for the subscription; this value is optional due to websocket subprotocol
• subscriptionType: The command type for the subscription; this value is optional due to websocket subprotocol, default is start, alernative value can be subscribe

addSubscriptionInit

addSubscriptionInit(clientId: string, payload): void

Registers the connection_init payload for a client.

Parameters:

• clientId: A unique identifier for the client
• payload: payload content

updateSubscriptionState

updateSubscriptionState(clientId: string, result): void

Updates the state of a client's subscription based on the latest result.

Parameters:

• clientId: The client's unique identifier
• result: The data object containing the subscription result

restoreSubscriptions

restoreSubscriptions(clientId: string, target): void

Restores all subscriptions for a client after reconnection.

Parameters:

• clientId: The client's unique identifier
• target: The WebSocket connection to send subscription requests to

removeSubscription

removeSubscription(clientId: string, subscriptionId: string): void

Removes the subscription for a specific client by subscription id.

Parameters:

• clientId: The client's unique identifier
• subscriptionId: A unique identifier for the subscription; note this value may not be present due to websocket message

removeAllSubscriptions

removeAllSubscriptions(clientId: string): void

Removes all state for a specific client, including its subscriptions and the connection_init payload. Use this when a client disconnects: the state is re-created lazily if the client sends new subscriptions afterwards.

Parameters:

• clientId: The client's unique identifier

Resume Logic

When a client reconnects, the library creates a recovery query using the last known value of the key field. The key field is automatically injected into the query parameters, even if it wasn't present in the original subscription query. For example, if a client was subscribed to updates with an offset of 42 before disconnecting, the recovery query will look like:

subscription {
  onItems(offset: 42) {
    id
    offset
    data
  }
}

If you've configured fixed arguments via the args property, these will be included in the recovery query as well:

subscription {
  onItems(offset: 42, filter: "important", limit: 10) {
    id
    offset
    data
  }
}

This query tells the GraphQL server to send updates starting from offset 42, ensuring the client doesn't miss any updates that occurred during the disconnection. The library handles this key injection automatically, so you don't need to modify your client-side subscription queries to support resumption.

Advanced Features

Subscription Aliases

The library fully supports GraphQL aliases. When a client uses an alias in their subscription:

subscription {
  itemsUpdates: onItems {
    id
    offset
    price
  }
}

The library will correctly track and restore the subscription using the alias.

Variable Handling

The library properly handles GraphQL variables in subscription queries:

subscription($lastOffset: Int!) {
  onItems(offset: $lastOffset) {
    id
    offset
    price
  }
}

With variables:

{
  "lastOffset": 100
}