Add experimental support for streaming SSE in live mode.

[thruflo]

If you add experimental_live_sse=true to your live requests, then the server streams SSEs rather than returning immediately when there's new data.

Requests are closed after 60 seconds, in order to support request collapsing. We also diverge slightly from default SSE behaviour by requiring the client to re-connect on a new URL once the request is closed. Because we require the client to honour our API mechanism of advancing the offset.

This can be worked around using a standard JS EventStream client by closing in the event of error and reconnecting manually. Just a rough sketch for reference:

const url = `http://localhost:3000/v1/shape?table=items&live=true&experimental_live_sse=true&handle=${handle}&offset=${offset}`
const es = new EventSource(url)
es.onerror = () => {
  es.close()
}

Note that the HTTP headers are returned at the start of the response. This means that the current header mechanism to return the next offset isn't valid. At the moment, you get a response like this:

curl -v 'http://localhost:3000/v1/shape?table=items&handle=22194952-1743787019364&live=true&experimental_live_sse=true&offset=5501319691220011464_0'
> GET /v1/shape?table=items&handle=22194952-1743787019364&live=true&experimental_live_sse=true&offset=5501319691220011464_0 HTTP/1.1
> Host: localhost:3000
> User-Agent: curl/8.7.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< date: Sun, 06 Apr 2025 11:12:45 GMT
< cache-control: public, max-age=59
< x-request-id: GDO2V0hAm4_Dp5cAABBj
< electric-server: ElectricSQL/1.0.5
< access-control-allow-origin: *
< access-control-expose-headers: *
< access-control-allow-methods: GET, HEAD, DELETE, OPTIONS
< content-type: text/event-stream
< electric-cursor: 15505980
< etag: "22194952-1743787019364:5501319691220011464_0:5501319691220011464_0"
< electric-handle: 22194952-1743787019364
< electric-up-to-date:
< electric-offset: 5501319691220011464_0
< connection: keep-alive
<
data: {"value":{"id":"9a33ad9e-39a6-4ab4-aebf-4a4c54c8dbc6"},"key":"\"public\".\"items\"/\"9a33ad9e-39a6-4ab4-aebf-4a4c54c8dbc6\"","headers":{"last":true,"relation":["public","items"],"operation":"insert","lsn":"5501319691220014840","op_position":0,"txids":[792]}}

data: {"headers":{"control":"up-to-date","global_last_seen_lsn":"5501319691220014840"}}

The global_last_seen_lsn is the correct lsn to resume from, so you can reconnect with e.g.: offset= 5501319691220014840_0 and it will continue streaming from the correct point.

I've tried to make sure that I keep everything a stream and don't either materialise or encode anything potentially expensive.

[netlify]

✅ Deploy Preview for electric-next ready!

Name	Link
🔨 Latest commit	5c3f5fc
🔍 Latest deploy log	https://app.netlify.com/projects/electric-next/deploys/685c119f0c48790008b66916
😎 Deploy Preview	https://deploy-preview-2544--electric-next.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[thruflo]

As per my previous discord note about diffing, the payload format of:

data: {"value":{"id":"9a33ad9e-39a6-4ab4-aebf-4a4c54c8dbc6"},"key":"\"public\".\"items\"/\"9a33ad9e-39a6-4ab4-aebf-4a4c54c8dbc6\"","headers":{"last":true,"relation":["public","items"],"operation":"insert","lsn":"5501319691220014840","op_position":0,"txids":[792]}}

data: {"headers":{"control":"up-to-date","global_last_seen_lsn":"5501319691220014840"}}

To send the smallest insert is not exactly optimal. Perhaps it gzips down ok over the wire to remove the duplication but it's possible to consider many approaches that could optimise down to an enum code and value, perhaps with a new offset at the end of the stream.

[icehaunter]

LGTM

[balegas]

If electric-offset is incorrect, should we use a special value when in SSE mode?

[kevin-dp]

If electric-offset is incorrect, should we use a special value when in SSE mode?

Perhaps even drop it from the headers?

[balegas]

yeah, but should also be clear that this is an SSE response from the headers (in case it isn't already) because this is a protocol change

[kevin-dp]

yeah, but should also be clear that this is an SSE response from the headers (in case it isn't already) because this is a protocol change

Agree. Could add a flag in the headers that indicates this is an SSE response.

[kevin-dp]

On a 409 the returned location header is wrong:

curl -i 'http://localhost:3000/v1/shape?table=foo&offset=26783992_0&handle=103702391-174791678028&live=true&experimental_live_sse=true'
HTTP/1.1 409 Conflict
date: Thu, 22 May 2025 12:32:23 GMT
content-length: 48
vary: accept-encoding
cache-control: public, max-age=60, must-revalidate
x-request-id: GEHZYbPCsu850hcAAAHB
electric-server: ElectricSQL/1.0.5
access-control-allow-origin: *
access-control-expose-headers: *
access-control-allow-methods: GET, HEAD, DELETE, OPTIONS
content-type: application/json; charset=utf-8
etag: "103702391-1747916780281::"
location: /v1/shape?experimental_live_sse=true&handle=103702391-1747916780281&offset=-1&table=foo
electric-handle: 103702391-1747916780281
electric-schema: {"a":{"type":"int4","not_null":true,"pk_index":0},"b":{"type":"int4"}}

data: [{"headers":{"control":"must-refetch"}}]

The location header contains the experimental_live_sse=true param but it does not include the live param, which is an invalid combination. So if i curl the returned location i get a 400:

curl -i 'http://localhost:3000/v1/shape?experimental_live_sse=true&handle=103702391-1747916780281&offset=-1&table=foo'
HTTP/1.1 400 Bad Request
date: Thu, 22 May 2025 12:33:29 GMT
content-length: 107
vary: accept-encoding
cache-control: no-cache
x-request-id: GEHZcSpbuSks7l4AAAIh
electric-server: ElectricSQL/1.0.5
access-control-allow-origin: *
access-control-expose-headers: *
access-control-allow-methods: GET, HEAD, DELETE, OPTIONS
content-type: application/json; charset=utf-8
electric-schema: null

{"message":"Invalid request","errors":{"experimental_live_sse":["can't be true unless live is also true"]}}

[kevin-dp]

Also, the data is always a JSON object, except on a 409, then it is an array:

curl -v "http://localhost:3000/v1/shape?cursor=&handle=62395453-1748244587988&live=true&offset=0_inf&table=electric_test.%22issues+for+181577045_0_3_0.ee54678b91ae8%22&experimental_live_sse=true"

data: {"value":{"id":"c4e5118d-4426-47cf-a509-27b05236d19d","priority":"10","title":"other title"},"key":"\"electric_test\".\"issues for 181577045_0_3_0.ee54678b91ae8\"/\"c4e5118d-4426-47cf-a509-27b05236d19d\"","headers":{"last":true,"relation":["electric_test","issues for 181577045_0_3_0.ee54678b91ae8"],"operation":"insert","lsn":"27322256","op_position":0,"txids":[807]}}

data: {"value":{"id":"4a36620f-2415-4c16-b41f-33cab1571f46","priority":"10","title":"other title2"},"key":"\"electric_test\".\"issues for 181577045_0_3_0.ee54678b91ae8\"/\"4a36620f-2415-4c16-b41f-33cab1571f46\"","headers":{"last":true,"relation":["electric_test","issues for 181577045_0_3_0.ee54678b91ae8"],"operation":"insert","lsn":"27322472","op_position":0,"txids":[808]}}

data: {"headers":{"control":"up-to-date","global_last_seen_lsn":"27322472"}}

Then i delete the shape, and curl again and i get a JSON array:

data: [{"headers":{"control":"must-refetch"}}]

[KyleAMathews]

Shipped in #2776

[kevin-dp]

I rebased this in #2856 and merged into main.