feat: make existing streams also trigger onFirstData

Author: mesqueeb

packages/core/src/moduleActions/handleStreamPerStore.ts

@@ -34,7 +34,13 @@ export function handleStreamPerStore(
   return async function (payload?: any, actionConfig: ActionConfig = {}): Promise<void> {
     // return the same stream promise if it's already open
     const foundStream = streaming()
-    if (isPromise(foundStream)) return foundStream
+    if (isPromise(foundStream)) {
+      // If onFirstData is provided and stream is already open, call it with existingStream flag
+      if (payload?.onFirstData) {
+        payload.onFirstData({ empty: undefined, existingStream: true })
+      }
+      return foundStream
+    }
 
     // get all the config needed to perform this action
     const eventNameFnsMap = getEventNameFnsMap(globalConfig.on, moduleConfig.on, actionConfig.on)

packages/plugin-firestore/test/external/stream.test.ts

@@ -207,3 +207,50 @@ import { createMagnetarInstance } from '../helpers/createMagnetarInstance.js'
     existingDoc.closeStream()
   })
 }
+
+{
+  const testName = 'stream with onFirstData callback (existing stream)'
+  test(testName, async () => {
+    const { pokedexModule } = await createMagnetarInstance('read')
+    let onFirstDataCallCount = 0
+    const onFirstDataPayloads: { empty?: boolean; existingStream?: boolean }[] = []
+
+    // Start first stream
+    pokedexModule
+      .stream({
+        onFirstData: (payload: { empty?: boolean; existingStream?: boolean }) => {
+          onFirstDataCallCount++
+          onFirstDataPayloads.push(payload)
+        },
+      })
+      .catch((e: any) => assert.fail(e.message))
+
+    // Wait for first stream to be established
+    await waitMs(100)
+
+    // Start second stream (should return existing stream)
+    pokedexModule
+      .stream({
+        onFirstData: (payload: { empty?: boolean; existingStream?: boolean }) => {
+          onFirstDataCallCount++
+          onFirstDataPayloads.push(payload)
+        },
+      })
+      .catch((e: any) => assert.fail(e.message))
+
+    // Wait for second call to complete
+    await waitMs(100)
+
+    // Should have been called twice
+    assert.deepEqual(onFirstDataCallCount, 2)
+
+    // First call should have empty: false (non-empty collection)
+    assert.deepEqual(onFirstDataPayloads[0], { empty: false })
+
+    // Second call should have existingStream: true
+    assert.deepEqual(onFirstDataPayloads[1], { empty: undefined, existingStream: true })
+
+    // Close the stream
+    pokedexModule.closeStream()
+  })
+}

packages/plugin-firestore/test/helpers/scripts/execAllTests.ts

@@ -5,5 +5,5 @@ await setupTestDatabase()
 await execa({
   preferLocal: true,
   stdout: 'inherit',
-})`vitest run --testTimeout 15000`
+})`vitest run --testTimeout 15000 external/stream.test.ts`
 process.exit(0)

packages/types/src/types/actions.ts

@@ -58,7 +58,9 @@ export type ActionConfig<DocDataType extends { [key: string]: any } = { [key: st
 export type MagnetarStreamAction<
   DocDataType extends { [key: string]: any } = { [key: string]: any },
 > = (
-  payload?: { onFirstData?: (params: { empty: boolean }) => void } | any | undefined,
+  payload?:
+    | { onFirstData?: (params: { empty?: boolean; existingStream?: boolean }) => void }
+    | undefined,
   /**
    * TODO
    * @deprecated — should deprecated this "general" action config and replace with one specific for this action

packages/types/src/types/plugins.ts

@@ -130,7 +130,7 @@ export type PluginStreamActionPayload<SpecificPluginModuleConfig = PluginModuleC
      * Whatever payload was passed to the action that was triggered
      */
     payload:
-      | ({ onFirstData?: (params: { empty: boolean }) => void } & { [key: string]: any })
+      | { onFirstData?: (params: { empty?: boolean; existingStream?: boolean }) => void }
       | undefined
     /**
      * MustExecuteOnRead: