Deriving the offset from the LSN in SSE mode is using a backend implementation detail that could change

[samwillis]

See: #2869 (review)

I don't like the existing getOffset method that constructs one from the LSN and a _0 suffix. The offset being constructed from an LSN is a backend implementation detail we should not depend on. I think we had this in my original POC as a work around, but we really should replace this by adding the actual offset on the up-to-date message, at least in SSE mode.

I would advocate for adding the actual offset to all up-to-date messages, on both SSE and long polling mode, it seems like a sensible place to place it.

[balegas]

We've had issues with building on the structure offset in the past, we should fix this. However, I'd say SSE is still experimental and not highest priority until we have more real usage of the feature.

We also need to be careful about adding anything to the up-to-date message since that would be a protocol that we'd have to maintain. So, let's discuss when it comes to making a decision.

[lekhnath]

I'm using live sse mode and could not figure out how do I get up to date offset after receiving new events through sse streaming response. Is it safe to add the _0 prefix to LSN manually or is there any other way around ?

[balegas]

What are you trying to achieve exactly? Seems this is a concern of the client.

[lekhnath]

@balegas Ok, I want to explain what I am trying to achieve.

I want to add real time streaming endpoints for my existing REST API in Django REST Framework. Here is how it works right now.

I already built an endpoint that takes an API path, resolves it to a viewset, gets the queryset, and from that determines the table name and the SQL WHERE condition. The WHERE condition is built dynamically from the URL query parameters using Django filters and the queryset.

For example:

/api/stream/v1/users?role=admin

This means:

v1/users is the API endpoint I want to stream.

role=admin becomes a SQL filter condition based on Django filters and the queryset.

Now for the Electric part.

I want to subscribe to SSE events using the Electric HTTP API and continuously send database changes back to the user. My current Python subscription code and the client usage are below.

The issue is this: I can only send the offset during the initial sync. I cannot get the offset during the live SSE events.

import logging
from typing import Any, Dict
from urllib.parse import urljoin

import httpx

from core.services.sse import stream_events

logger = logging.getLogger()


class ElectricClient:
    def __init__(self, base_url: str, headers: Dict[str, Any] = {}, secret: str | None = None):
        self.base_url = base_url

        self.headers = headers
        self.secret = secret
        self.client = httpx.AsyncClient(timeout=httpx.Timeout(5.0, read=None))

        self._shape_url = urljoin(self.base_url, "/v1/shape")

    @property
    def shape_url(self) -> str:
        return self._shape_url

    def params_from_response(self, response: httpx.Response) -> Dict[str, None]:
        return {
            "offset": response.headers.get("electric-offset"),
            "handle": response.headers.get("electric-handle"),
            "cursor": response.headers.get("electric-cursor"),
            "etag": response.headers.get("etag"),
            "up_to_date": "electric-up-to-date" in response.headers,
        }

    async def sync_changes(self, base_params: Dict[str, Any], offset: str = "now", handle: str | None = None):
        shape_logs = []

        def is_up_to_date(header_params, data):
            if header_params["up_to_date"]:
                return True

            up_to_date = False
            for rec in data:
                control = (rec.get("headers") or {}).get("control", None)

                if control == "up-to-date":
                    up_to_date = True
                    break

            return up_to_date

        while True:
            resp = await self.client.get(
                self.shape_url, params=dict(base_params, offset=offset, handle=handle), headers=self.headers
            )

            try:
                resp.raise_for_status()

                header_params = self.params_from_response(resp)

                data = resp.json()

                offset = header_params["offset"]
                handle = header_params["handle"]
                shape_logs += data

                if is_up_to_date(header_params, data):
                    break
            except httpx.HTTPStatusError:
                if offset == "now":
                    handle = None
                    break

                offset = "now"
                handle = None

        return {
            "offset": offset,
            "handle": handle,
            "shape_logs": shape_logs,
        }

    async def subscribe(
        self, table: str, offset: str = "now", handle: str | None = None, replica: str = "default", extra_params={}
    ):
        base_params = {
            "table": table,
            "replica": replica,
            "secret": self.secret,
            "log": "changes_only",
            **extra_params,
        }

        synced = await self.sync_changes(
            base_params,
            offset=offset,
            handle=handle,
        )

        yield {
            "event": "message",
            "data": {
                "headers": {
                    "control": "synced",
                    **synced,
                }
            },
        }

        for evt in synced["shape_logs"]:
            yield {"event": "message", "data": evt, "id": None, "retry": None}

        live_params = dict(
            base_params,
            **{
                "live": "true",
                "live_sse": "true",
                "offset": synced["offset"],
                "handle": synced["handle"],
            },
        )

        async for sse in stream_events(self.shape_url, params=live_params, client=self.client):
            yield sse

react hook (typescript)

import { fetchEventSource, EventStreamContentType } from '@microsoft/fetch-event-source';
import React from "react";

import { useAuthentication } from "desk/auth/providers/AuthenticationProvider";
import { trimSlashes, trimSlashRight } from 'desk/utils/extra';

class RetriableError extends Error {
    constructor(message: string, public retryAfter: number = 1000) {
        super(message);
        this.retryAfter = retryAfter;
    }
}

class FatalError extends Error { }

type StreamEvent = {
    event: string;
    data: any;
}
export type StreamEventHandler = (event: StreamEvent) => void;

const streamLastEventMap = new Map<string, string>();

export default function useStream(apiEndpoint: string | undefined, queryParams: URLSearchParams, onEvent: StreamEventHandler, deps: any[] = []) {
    const auth = useAuthentication();
    const abortController = React.useRef<AbortController | null>(null);

    React.useEffect(() => {
        if (abortController.current) {
            abortController.current.abort();
        }

        abortController.current = new AbortController();

        if (!apiEndpoint) { return; }

        // console.log('sending...');

        const search = new URLSearchParams(queryParams);

        if (!search.has("page")) {
            search.set("page", "1");
        }

        const baseUrl = trimSlashRight(import.meta.env.APP_API_ENDPOINT);
        const resourceUrl = trimSlashes(apiEndpoint);
        const fetchUrl = `${baseUrl}/v1/stream/${resourceUrl}/?${search.toString()}`;
        const lastEventId = streamLastEventMap.get(fetchUrl);

        // console.log('--last event id = ', lastEventId);

        fetchEventSource(fetchUrl, {
            method: "GET",
            openWhenHidden: true,
            signal: abortController.current.signal,
            headers: {
                'Content-Type': 'application/json',
                'Accept': EventStreamContentType,
                'Authorization': auth.accessToken ? `Bearer ${auth.accessToken}` : '',
                'Last-Event-ID': lastEventId ? lastEventId : '',
            },
            async onopen(response) {
                const ct = response.headers.get('content-type') || '';
                if (response.ok && ct.includes(EventStreamContentType)) {
                    return;
                };

                // 4xx (except 429) "don't retry"
                if (response.status >= 400 && response.status < 500 && response.status !== 429) {
                    throw new FatalError(`Fatal HTTP ${response.status}`);
                }

                // 5xx/429/etc: retry
                throw new RetriableError(`Retry HTTP ${response.status}`);
            },
            onmessage(message) {
                if (message.id) {
                    // console.log('--set new event id ', fetchUrl, message.id);
                    streamLastEventMap.set(fetchUrl, message.id);
                }

                const messageEvent = message.event;
                const messageData = JSON.parse(message.data || '{}');

                if (messageEvent == "control" && messageData.type === "reconnect") {
                    throw new RetriableError("Retry", message.retry || 1000)
                }

                onEvent({ event: messageEvent, data: messageData }); // TODO: update api resource query cache also after receiving realtime api data?
            },
            onclose() {
                console.log('-- streaming connection closed');
            },
            onerror(error) {
                if (error instanceof RetriableError) {
                    return error.retryAfter || 1000;
                } else {
                    // signals.onResponse({ "error": error.message });// displays server error message
                    console.error('--streaming error', error);
                    throw error; // rethrow to stop the operation
                }
            }

        });

        return () => {
            if (abortController.current) {
                abortController.current.abort();
                abortController.current = null;
            }
        }

    }, [apiEndpoint, auth, queryParams, ...deps]);

}