feat(ts-client): Start stream after subscribe (#2188)

Addresses part of #1997

Had to change some tests around - obviously this is not a full move
towards the desired behaviour, e.g. we might even want for the stream to
stop when it has _no_ subscribers, but I think this provides some
beneficial properties to the stream (e.g. if subscribing to it after
async gap no messages lost)

If we think this just complicates the client and that if we want to
address the issue we should do it in one comprehensive go, I'm happy to
close this as well

.changeset/mean-humans-clean.md

@@ -0,0 +1,5 @@
+---
+"@electric-sql/client": patch
+---
+
+Start streaming only after at least one subscriber is present.

packages/typescript-client/src/client.ts

@@ -257,6 +257,7 @@ export class ShapeStream<T extends Row<unknown> = Row>
     ]
   >()
 
+  #started = false
   #lastOffset: Offset
   #liveCacheBuster: string // Seconds since our Electric Epoch 😎
   #lastSyncedAt?: number // unix time
@@ -290,8 +291,6 @@ export class ShapeStream<T extends Row<unknown> = Row>
     this.#fetchClient = createFetchWithResponseHeadersCheck(
       createFetchWithChunkBuffer(fetchWithBackoffClient)
     )
-
-    this.#start()
   }
 
   get shapeHandle() {
@@ -311,6 +310,9 @@ export class ShapeStream<T extends Row<unknown> = Row>
   }
 
   async #start() {
+    if (this.#started) throw new Error(`Cannot start stream twice`)
+    this.#started = true
+
     try {
       while (
         (!this.options.signal?.aborted && !this.#isUpToDate) ||
@@ -462,6 +464,7 @@ export class ShapeStream<T extends Row<unknown> = Row>
           }
 
           // Restart
+          this.#started = false
           this.#start()
         }
         return
@@ -481,6 +484,7 @@ export class ShapeStream<T extends Row<unknown> = Row>
     const subscriptionId = Math.random()
 
     this.#subscribers.set(subscriptionId, [callback, onError])
+    if (!this.#started) this.#start()
 
     return () => {
       this.#subscribers.delete(subscriptionId)

packages/typescript-client/test/client.test.ts

@@ -282,7 +282,8 @@ describe(`Shape`, () => {
       signal: aborter.signal,
     })
 
-    await sleep(100) // give some time for the initial fetch to complete
+    // give some time for the initial fetch to complete
+    await waitForFetch(shapeStream)
     expect(shapeStream.isConnected()).true
 
     const shape = new Shape(shapeStream)
@@ -328,6 +329,8 @@ describe(`Shape`, () => {
       },
     })
 
+    const unsubscribe = shapeStream.subscribe(() => unsubscribe())
+
     await sleep(100) // give some time for the initial fetch to complete
     expect(shapeStream.isConnected()).true
 
@@ -347,7 +350,7 @@ describe(`Shape`, () => {
     issuesTableUrl,
   }) => {
     const mockErrorHandler = vi.fn()
-    new ShapeStream({
+    const shapeStream = new ShapeStream({
       url: `${BASE_URL}/v1/shape`,
       params: {
         table: issuesTableUrl,
@@ -360,7 +363,7 @@ describe(`Shape`, () => {
       onError: mockErrorHandler,
     })
 
-    await sleep(10) // give some time for the initial fetch to complete
+    await waitForFetch(shapeStream)
     expect(mockErrorHandler.mock.calls.length).toBe(1)
     expect(mockErrorHandler.mock.calls[0][0]).toBeInstanceOf(FetchError)
   })
@@ -383,7 +386,7 @@ describe(`Shape`, () => {
       }
     })
 
-    new ShapeStream({
+    const shapeStream = new ShapeStream({
       url: `${BASE_URL}/v1/shape`,
       params: {
         table: issuesTableUrl,
@@ -402,7 +405,7 @@ describe(`Shape`, () => {
       onError: mockErrorHandler,
     })
 
-    await sleep(50) // give some time for the fetches to complete
+    await waitForFetch(shapeStream)
     expect(mockErrorHandler.mock.calls.length).toBe(1)
     expect(mockErrorHandler.mock.calls[0][0]).toBeInstanceOf(FetchError)
   })
@@ -425,7 +428,7 @@ describe(`Shape`, () => {
       }
     })
 
-    new ShapeStream({
+    const shapeStream = new ShapeStream({
       url: `${BASE_URL}/v1/shape`,
       params: {
         table: issuesTableUrl,
@@ -446,7 +449,7 @@ describe(`Shape`, () => {
       onError: mockErrorHandler,
     })
 
-    await sleep(50) // give some time for the fetches to complete
+    await waitForFetch(shapeStream)
     expect(mockErrorHandler.mock.calls.length).toBe(1)
     expect(mockErrorHandler.mock.calls[0][0]).toBeInstanceOf(FetchError)
   })
@@ -489,7 +492,7 @@ describe(`Shape`, () => {
       onError: mockErrorHandler,
     })
 
-    await sleep(50) // give some time for the first fetch to complete
+    await waitForFetch(shapeStream)
     expect(mockErrorHandler.mock.calls.length).toBe(1)
     expect(mockErrorHandler.mock.calls[0][0]).toBeInstanceOf(FetchError)
     expect(shapeStream.isConnected()).toBe(false)
@@ -521,7 +524,9 @@ describe(`Shape`, () => {
       },
     })
 
-    await sleep(10) // give some time for the initial fetch to complete
+    const unsub = shapeStream.subscribe(() => unsub())
+    await sleep(10) // give sometime for fetch to fail
+
     expect(shapeStream.isConnected()).false
 
     const expectedErrorMessage = new MissingHeadersError(url, [
@@ -549,6 +554,7 @@ describe(`Shape`, () => {
       },
     })
 
+    const unsubLive = shapeStreamLive.subscribe(() => unsubLive())
     await sleep(10) // give some time for the initial fetch to complete
     expect(shapeStreamLive.isConnected()).false
 
@@ -573,7 +579,8 @@ describe(`Shape`, () => {
       subscribe: false,
     })
 
-    await sleep(100) // give some time for the fetch to complete
+    await waitForFetch(shapeStream)
+    await sleep(50)
 
     // We should no longer be connected because
     // the initial fetch finished and we've not subscribed to changes
@@ -594,7 +601,7 @@ describe(`Shape`, () => {
 
     expect(shapeStream.isLoading()).true
 
-    await sleep(200) // give some time for the initial fetch to complete
+    await waitForFetch(shapeStream)
 
     expect(shapeStream.isLoading()).false
   })
@@ -665,3 +672,13 @@ describe(`Shape`, () => {
     }
   })
 })
+
+function waitForFetch(stream: ShapeStream): Promise<void> {
+  let unsub = () => {}
+  return new Promise<void>((resolve) => {
+    unsub = stream.subscribe(
+      () => resolve(),
+      () => resolve()
+    )
+  }).finally(() => unsub())
+}

packages/typescript-client/test/stream.test.ts

@@ -23,7 +23,7 @@ describe(`ShapeStream`, () => {
     }
 
     const aborter = new AbortController()
-    new ShapeStream({
+    const stream = new ShapeStream({
       url: shapeUrl,
       params: {
         table: `foo`,
@@ -35,6 +35,7 @@ describe(`ShapeStream`, () => {
         'X-Custom-Header': `my-value`,
       },
     })
+    const unsub = stream.subscribe(() => unsub())
 
     await new Promise((resolve) =>
       eventTarget.addEventListener(`fetch`, resolve, { once: true })
@@ -60,7 +61,7 @@ describe(`ShapeStream`, () => {
     }
 
     const aborter = new AbortController()
-    new ShapeStream({
+    const stream = new ShapeStream({
       url: shapeUrl,
       params: {
         table: `foo`,
@@ -72,6 +73,8 @@ describe(`ShapeStream`, () => {
       fetchClient: fetchWrapper,
     })
 
+    const unsub = stream.subscribe(() => unsub())
+
     await new Promise((resolve) =>
       eventTarget.addEventListener(`fetch`, resolve, { once: true })
     )
@@ -80,4 +83,41 @@ describe(`ShapeStream`, () => {
       `columns=id&handle=potato&offset=-1&table=foo&where=a%3D1`
     )
   })
+
+  it(`should start requesting only after first subscription`, async () => {
+    const eventTarget = new EventTarget()
+    const fetchWrapper = (): Promise<Response> => {
+      eventTarget.dispatchEvent(new Event(`fetch`))
+      return Promise.resolve(Response.error())
+    }
+
+    const aborter = new AbortController()
+    const stream = new ShapeStream({
+      url: shapeUrl,
+      params: {
+        table: `foo`,
+        where: `a=1`,
+        columns: [`id`],
+      },
+      handle: `potato`,
+      signal: aborter.signal,
+      fetchClient: fetchWrapper,
+    })
+
+    // should not fire any fetch requests
+    await new Promise<void>((resolve, reject) => {
+      eventTarget.addEventListener(`fetch`, reject, { once: true })
+      setTimeout(() => resolve(), 100)
+    })
+
+    // should fire fetch immediately after subbing
+    const startedStreaming = new Promise<void>((resolve, reject) => {
+      eventTarget.addEventListener(`fetch`, () => resolve(), {
+        once: true,
+      })
+      setTimeout(() => reject(`timed out`), 100)
+    })
+    const unsub = stream.subscribe(() => unsub())
+    await startedStreaming
+  })
 })