From 25643819690ad6ce98ff948f9c7f245a87d7aa47 Mon Sep 17 00:00:00 2001
From: Ilia Borovitinov <ilia@electric-sql.com>
Date: Thu, 11 Dec 2025 13:50:26 +0300
Subject: [PATCH] feat: denote end of injected subset snapshot with an
 additional message

---
 .changeset/swift-brooms-double.md         |  5 +++++
 packages/typescript-client/src/client.ts  | 10 ++--------
 packages/typescript-client/src/helpers.ts |  6 ++----
 packages/typescript-client/src/types.ts   |  9 +++++++++
 4 files changed, 18 insertions(+), 12 deletions(-)
 create mode 100644 .changeset/swift-brooms-double.md

diff --git a/.changeset/swift-brooms-double.md b/.changeset/swift-brooms-double.md
new file mode 100644
index 0000000000..c36a163c39
--- /dev/null
+++ b/.changeset/swift-brooms-double.md
@@ -0,0 +1,5 @@
+---
+'@electric-sql/client': minor
+---
+
+feat: denote end of injected subset snapshot with an additional message
diff --git a/packages/typescript-client/src/client.ts b/packages/typescript-client/src/client.ts
index fa5457e75d..04edff6a55 100644
--- a/packages/typescript-client/src/client.ts
+++ b/packages/typescript-client/src/client.ts
@@ -7,6 +7,7 @@ import {
   GetExtensions,
   ChangeMessage,
   SnapshotMetadata,
+  SubsetParams,
 } from './types'
 import { MessageParser, Parser, TransformFunction } from './parser'
 import { ColumnMapper, encodeWhereClause } from './column-mapper'
@@ -128,14 +129,6 @@ export type ExternalParamsRecord<T extends Row<unknown> = Row> = {
   [K in string]: ParamValue | undefined
 } & Partial<PostgresParams<T>> & { [K in ReservedParamKeys]?: never }
 
-export type SubsetParams = {
-  where?: string
-  params?: Record<string, string>
-  limit?: number
-  offset?: number
-  orderBy?: string
-}
-
 type ReservedParamKeys =
   | typeof LIVE_CACHE_BUSTER_QUERY_PARAM
   | typeof SHAPE_HANDLE_QUERY_PARAM
@@ -1457,6 +1450,7 @@ export class ShapeStream<T extends Row<unknown> = Row>
 
       const dataWithEndBoundary = (data as Array<Message<T>>).concat([
         { headers: { control: `snapshot-end`, ...metadata } },
+        { headers: { control: `subset-end`, ...opts } },
       ])
 
       this.#snapshotTracker.addSnapshot(
diff --git a/packages/typescript-client/src/helpers.ts b/packages/typescript-client/src/helpers.ts
index 6cfcbb2cd8..c5a5604e4d 100644
--- a/packages/typescript-client/src/helpers.ts
+++ b/packages/typescript-client/src/helpers.ts
@@ -66,11 +66,9 @@ export function isUpToDateMessage<T extends Row<unknown> = Row>(
  * If we are not in SSE mode this function will return undefined.
  */
 export function getOffset(message: ControlMessage): Offset | undefined {
+  if (message.headers.control != `up-to-date`) return
   const lsn = message.headers.global_last_seen_lsn
-  if (!lsn) {
-    return
-  }
-  return `${lsn}_0` as Offset
+  return lsn ? (`${lsn}_0` as Offset) : undefined
 }
 
 /**
diff --git a/packages/typescript-client/src/types.ts b/packages/typescript-client/src/types.ts
index 01a6a0c1bb..d9eac3d20d 100644
--- a/packages/typescript-client/src/types.ts
+++ b/packages/typescript-client/src/types.ts
@@ -68,6 +68,14 @@ export type MoveTag = string
  */
 export type MoveOutPattern = { pos: number; value: string }
 
+export type SubsetParams = {
+  where?: string
+  params?: Record<string, string>
+  limit?: number
+  offset?: number
+  orderBy?: string
+}
+
 export type ControlMessage = {
   headers:
     | (Header & {
@@ -75,6 +83,7 @@ export type ControlMessage = {
         global_last_seen_lsn?: string
       })
     | (Header & { control: `snapshot-end` } & PostgresSnapshot)
+    | (Header & { control: `subset-end` } & SubsetParams)
 }
 
 export type EventMessage = {