fix(pglite-sync): don't filter move-in messages by LSN (#883)

* fix(pglite-sync): don't filter move-in messages by LSN

Move-in messages from Electric's tagged_subqueries feature don't have
an LSN header because they come from direct DB queries, not replication.
Previously these messages were incorrectly skipped as "already seen"
because the missing LSN defaulted to 0.

This checks for the is_move_in header and bypasses LSN filtering for
move-in messages, ensuring rows moving into a shape due to subquery
condition changes are properly synced.

Fixes electric-sql/electric#3769

* fix(pglite-sync): handle move-in duplicate keys with ON CONFLICT

Move-in data from tagged_subqueries can overlap with initial sync data,
causing duplicate key errors. This adds ON CONFLICT DO UPDATE handling
specifically for move-in inserts.

- Add primaryKey param to applyInsertsToTable
- Use ON CONFLICT DO UPDATE for move-in inserts
- Update changeset description

* Add regression tests for move-in messages

* Fix formatting

---------

Co-authored-by: msfstef <msfstef@gmail.com>

Author: jbingen

.changeset/fix-move-in-lsn-filtering.md

@@ -0,0 +1,11 @@
+---
+"@electric-sql/pglite-sync": patch
+---
+
+Fix move-in messages from tagged_subqueries not being synced
+
+This fixes two issues with move-in messages from Electric's `tagged_subqueries` feature:
+
+1. **LSN filtering bypass**: Move-in messages don't include an LSN header because they originate from direct database queries rather than the PostgreSQL replication stream. Previously, these messages were being filtered out as "already seen" because the missing LSN defaulted to 0. This fix checks for the `is_move_in` header and bypasses LSN filtering for these messages.
+
+2. **Duplicate key handling**: Move-in data can overlap with data from the initial sync (e.g., when a row "moves in" to match a subquery that it already matched during initial sync). This fix uses `ON CONFLICT DO UPDATE` for move-in inserts to handle these duplicates gracefully, updating the row with the latest data instead of erroring.

packages/pglite-sync/src/apply.ts

@@ -23,16 +23,39 @@ export async function applyMessageToTable({
 }: ApplyMessageToTableOptions) {
   const data = mapColumns ? doMapColumns(mapColumns, message) : message.value
 
+  // Check if this is a move-in message (from subquery-based shapes)
+  const isMoveIn =
+    (message.headers as Record<string, unknown>).is_move_in === true
+
   switch (message.headers.operation) {
     case 'insert': {
       if (debug) console.log('inserting', data)
       const columns = Object.keys(data)
+
+      // Build ON CONFLICT clause for move-in messages
+      let onConflictClause = ''
+      if (isMoveIn && primaryKey && primaryKey.length > 0) {
+        const nonPkColumns = columns.filter((c) => !primaryKey.includes(c))
+        if (nonPkColumns.length > 0) {
+          onConflictClause = `
+            ON CONFLICT (${primaryKey.map((c) => `"${c}"`).join(', ')})
+            DO UPDATE SET ${nonPkColumns.map((c) => `"${c}" = EXCLUDED."${c}"`).join(', ')}
+          `
+        } else {
+          onConflictClause = `
+            ON CONFLICT (${primaryKey.map((c) => `"${c}"`).join(', ')})
+            DO NOTHING
+          `
+        }
+      }
+
       return await pg.query(
         `
             INSERT INTO "${schema}"."${table}"
             (${columns.map((s) => '"' + s + '"').join(', ')})
             VALUES
             (${columns.map((_v, i) => '$' + (i + 1)).join(', ')})
+            ${onConflictClause}
           `,
         columns.map((column) => data[column]),
       )
@@ -86,6 +109,7 @@ export interface BulkApplyMessagesToTableOptions {
   schema?: string
   messages: InsertChangeMessage[]
   mapColumns?: MapColumns
+  primaryKey?: string[]
   debug: boolean
 }
 
@@ -95,6 +119,7 @@ export async function applyInsertsToTable({
   schema = 'public',
   messages,
   mapColumns,
+  primaryKey,
   debug,
 }: BulkApplyMessagesToTableOptions) {
   // Map the messages to the data to be inserted
@@ -104,6 +129,12 @@ export async function applyInsertsToTable({
 
   if (debug) console.log('inserting', data)
 
+  // Check if any message is a move-in (from subquery-based shapes)
+  // Move-in data can overlap with existing data, so we need ON CONFLICT handling
+  const hasMoveIn = messages.some(
+    (m) => (m.headers as Record<string, unknown>).is_move_in === true,
+  )
+
   // Get column names from the first message
   const columns = Object.keys(data[0])
 
@@ -176,11 +207,31 @@ export async function applyInsertsToTable({
 
   // Helper function to execute a batch insert
   const executeBatch = async (batch: Record<string, any>[]) => {
+    // Build ON CONFLICT clause for move-in messages
+    // Move-in data can contain rows that already exist from initial sync
+    let onConflictClause = ''
+    if (hasMoveIn && primaryKey && primaryKey.length > 0) {
+      const nonPkColumns = columns.filter((c) => !primaryKey.includes(c))
+      if (nonPkColumns.length > 0) {
+        onConflictClause = `
+          ON CONFLICT (${primaryKey.map((c) => `"${c}"`).join(', ')})
+          DO UPDATE SET ${nonPkColumns.map((c) => `"${c}" = EXCLUDED."${c}"`).join(', ')}
+        `
+      } else {
+        // All columns are primary key, just ignore duplicates
+        onConflictClause = `
+          ON CONFLICT (${primaryKey.map((c) => `"${c}"`).join(', ')})
+          DO NOTHING
+        `
+      }
+    }
+
     const sql = `
       INSERT INTO "${schema}"."${table}"
       (${columns.map((s) => `"${s}"`).join(', ')})
       VALUES
       ${batch.map((_, j) => `(${columns.map((_v, k) => '$' + (j * columns.length + k + 1)).join(', ')})`).join(', ')}
+      ${onConflictClause}
     `
     const values = batch.flatMap((message) =>
       columns.map((column) => message[column]),

packages/pglite-sync/src/index.ts

@@ -255,6 +255,7 @@ async function createPlugin(
                 schema: shape.schema,
                 messages: initialInserts as InsertChangeMessage[],
                 mapColumns: shape.mapColumns,
+                primaryKey: shape.primaryKey,
                 debug,
               })
 
@@ -282,6 +283,7 @@ async function createPlugin(
                   schema: shape.schema,
                   messages: bulkInserts as InsertChangeMessage[],
                   mapColumns: shape.mapColumns,
+                  primaryKey: shape.primaryKey,
                   debug,
                 })
                 bulkInserts.length = 0
@@ -351,7 +353,14 @@ async function createPlugin(
             typeof message.headers.lsn === 'string'
               ? BigInt(message.headers.lsn)
               : BigInt(0) // we default to 0 if there no lsn on the message
-          if (lsn <= lastCommittedLsnForShape) {
+
+          // Move-in messages from subquery-based shapes don't have an LSN
+          // because they come from direct DB queries, not from replication.
+          // We should never skip these based on LSN filtering.
+          const isMoveIn =
+            (message.headers as Record<string, unknown>).is_move_in === true
+
+          if (!isMoveIn && lsn <= lastCommittedLsnForShape) {
             // We are replaying changes / have already seen this lsn
             // skip and move on to the next message
             return

packages/pglite-sync/test/sync.test.ts

@@ -172,6 +172,133 @@ describe('pglite-sync', () => {
     shape.unsubscribe()
   })
 
+  it('syncs move-in messages that have no lsn', async () => {
+    let feedMessage: (
+      lsn: number,
+      message: MultiShapeMessage,
+    ) => Promise<void> = async (_) => {}
+    MockMultiShapeStream.mockImplementation(() => ({
+      subscribe: vi.fn(
+        (cb: (messages: MultiShapeMessage[]) => Promise<void>) => {
+          feedMessage = (lsn, message) =>
+            cb([
+              message,
+              {
+                shape: 'shape',
+                headers: {
+                  control: 'up-to-date',
+                  global_last_seen_lsn: lsn.toString(),
+                },
+              },
+            ])
+        },
+      ),
+      unsubscribeAll: vi.fn(),
+      isUpToDate: true,
+      shapes: {
+        shape: {
+          subscribe: vi.fn(),
+          unsubscribeAll: vi.fn(),
+        },
+      },
+    }))
+
+    const shape = await pg.electric.syncShapeToTable({
+      shape: {
+        url: 'http://localhost:3000/v1/shape',
+        params: { table: 'todo' },
+      },
+      table: 'todo',
+      primaryKey: ['id'],
+      shapeKey: null,
+    })
+
+    // initial sync, commits past lsn 0
+    await feedMessage(5, {
+      headers: { operation: 'insert', lsn: '5' },
+      key: 'id1',
+      value: { id: 1, task: 'task1', done: false },
+      shape: 'shape',
+    })
+
+    // move-in has no lsn (defaults to 0), must not be skipped as already seen
+    await feedMessage(6, {
+      headers: { operation: 'insert', is_move_in: true },
+      key: 'id2',
+      value: { id: 2, task: 'task2', done: false },
+      shape: 'shape',
+    })
+    expect((await pg.sql`SELECT * FROM todo ORDER BY id;`).rows).toEqual([
+      { id: 1, task: 'task1', done: false },
+      { id: 2, task: 'task2', done: false },
+    ])
+
+    shape.unsubscribe()
+  })
+
+  it('upserts move-in messages that overlap with existing rows', async () => {
+    let feedMessage: (
+      lsn: number,
+      message: MultiShapeMessage,
+    ) => Promise<void> = async (_) => {}
+    MockMultiShapeStream.mockImplementation(() => ({
+      subscribe: vi.fn(
+        (cb: (messages: MultiShapeMessage[]) => Promise<void>) => {
+          feedMessage = (lsn, message) =>
+            cb([
+              message,
+              {
+                shape: 'shape',
+                headers: {
+                  control: 'up-to-date',
+                  global_last_seen_lsn: lsn.toString(),
+                },
+              },
+            ])
+        },
+      ),
+      unsubscribeAll: vi.fn(),
+      isUpToDate: true,
+      shapes: {
+        shape: {
+          subscribe: vi.fn(),
+          unsubscribeAll: vi.fn(),
+        },
+      },
+    }))
+
+    const shape = await pg.electric.syncShapeToTable({
+      shape: {
+        url: 'http://localhost:3000/v1/shape',
+        params: { table: 'todo' },
+      },
+      table: 'todo',
+      primaryKey: ['id'],
+      shapeKey: null,
+    })
+
+    // initial sync
+    await feedMessage(5, {
+      headers: { operation: 'insert', lsn: '5' },
+      key: 'id1',
+      value: { id: 1, task: 'task1', done: false },
+      shape: 'shape',
+    })
+
+    // move-in with an existing key must upsert, not error on duplicate key
+    await feedMessage(6, {
+      headers: { operation: 'insert', is_move_in: true },
+      key: 'id1',
+      value: { id: 1, task: 'task1-updated', done: true },
+      shape: 'shape',
+    })
+    expect((await pg.sql`SELECT * FROM todo;`).rows).toEqual([
+      { id: 1, task: 'task1-updated', done: true },
+    ])
+
+    shape.unsubscribe()
+  })
+
   it('performs operations within a transaction', async () => {
     let feedMessages: (
       lsn: number,