make sure it's called _last_syned_at in all places

Author: Yostra

e2e/stripe-reconcile-cleanup.test.ts

@@ -125,7 +125,7 @@ describeWithEnv(
 
       try {
         // Backfill-only sync (no websocket, no event polling) — both rows
-        // land in postgres with `_synced_at ≈ T0`.
+        // land in postgres with `_last_synced_at ≈ T0`.
         await drain(engine.pipeline_sync(pipeline))
 
         const seeded = await pool.query<{ id: string }>(
@@ -139,8 +139,8 @@ describeWithEnv(
         await stripe.customers.del(doomed.id)
         cleanupIds.delete(doomed.id)
 
-        // `_synced_at` is set with millisecond precision by the destination,
-        // so a small forward skew guarantees `syncRunStartedAt > _synced_at`.
+        // `_last_synced_at` is set with millisecond precision by the destination,
+        // so a small forward skew guarantees `syncRunStartedAt > _last_synced_at`.
         await new Promise((r) => setTimeout(r, 50))
         const syncRunStartedAt = new Date().toISOString()
 
@@ -268,7 +268,7 @@ describeWithEnv(
       const cleanupIds = new Set<string>([survivor.id, doomed.id])
 
       try {
-        // Backfill seeds both customers with `_synced_at ≈ T0`.
+        // Backfill seeds both customers with `_last_synced_at ≈ T0`.
         await drain(engine.pipeline_sync(pipeline))
 
         const seededRows = await readSheet(sheetsClient, spreadsheetId, STREAM)

packages/destination-google-sheets/__tests__/integration.test.ts

@@ -27,7 +27,7 @@ describeWithEnv(
     function stripUpdatedAt(rows: unknown[][]): unknown[][] {
       const header = rows[0] ?? []
       const indexes = new Set(
-        ['_updated_at', '_synced_at'].map((name) => header.indexOf(name)).filter((idx) => idx >= 0)
+        ['_updated_at', '_last_synced_at'].map((name) => header.indexOf(name)).filter((idx) => idx >= 0)
       )
       if (indexes.size === 0) return rows
       return rows.map((row) => row.filter((_, i) => !indexes.has(i)))

packages/destination-google-sheets/src/index.test.ts

@@ -20,14 +20,14 @@ import { createMemorySheets } from '../__tests__/memory-sheets.js'
 /**
  * Strip metadata timestamp columns from a 2D rows array.
  *
- * The destination stamps `_synced_at`; the source may stamp `_updated_at`.
+ * The destination stamps `_last_synced_at`; the source may stamp `_updated_at`.
  * Most tests only care about source data, so drop both at assertion sites.
  */
 function stripUpdatedAt(rows: unknown[][] | undefined): unknown[][] {
   if (!rows || rows.length === 0) return rows ?? []
   const header = rows[0] as unknown[]
   const indexes = new Set(
-    ['_updated_at', '_synced_at'].map((name) => header.indexOf(name)).filter((idx) => idx >= 0)
+    ['_updated_at', '_last_synced_at'].map((name) => header.indexOf(name)).filter((idx) => idx >= 0)
   )
   if (indexes.size === 0) return rows
   return rows.map((row) => row.filter((_, i) => !indexes.has(i)))
@@ -1786,9 +1786,9 @@ describe('_updated_at column (source-owned, passthrough)', () => {
     )
 
     const rows = getData(getSpreadsheetIds()[0], 'users')!
-    expect(rows[0]).toEqual(['id', 'name', '_synced_at'])
+    expect(rows[0]).toEqual(['id', 'name', '_last_synced_at'])
     expect(rows[0]).not.toContain('_updated_at')
-    const syncedAtIdx = (rows[0] as string[]).indexOf('_synced_at')
+    const syncedAtIdx = (rows[0] as string[]).indexOf('_last_synced_at')
     expect(syncedAtIdx).toBeGreaterThanOrEqual(0)
     expect(Date.parse(String(rows[1][syncedAtIdx]))).not.toBeNaN()
   })
@@ -2062,7 +2062,7 @@ describe('getStaleRecords', () => {
     ],
   }
 
-  it('returns ids whose _synced_at predates syncRunStartedAt', async () => {
+  it('returns ids whose _last_synced_at predates syncRunStartedAt', async () => {
     const { sheets, getSpreadsheetIds } = createMemorySheets()
     const dest = createDestination(sheets)
 
@@ -2091,7 +2091,7 @@ describe('getStaleRecords', () => {
     expect(batches).toEqual([{ stream: 'customer', ids: ['cus_1', 'cus_2'] }])
   })
 
-  it('does not return rows whose _synced_at is at or after syncRunStartedAt', async () => {
+  it('does not return rows whose _last_synced_at is at or after syncRunStartedAt', async () => {
     const { sheets, getSpreadsheetIds } = createMemorySheets()
     const dest = createDestination(sheets)
 

packages/destination-google-sheets/src/index.ts

@@ -846,10 +846,14 @@ export function createDestination(
               }
             }
 
-            const headers = await ensureHeadersForRecord(stream, cleanData)
+            const syncedData: Record<string, unknown> = {
+              ...cleanData,
+              _last_synced_at: new Date().toISOString(),
+            }
+            const headers = await ensureHeadersForRecord(stream, syncedData)
             const tsFields = streamTimestampFields.get(stream)
             const row = headers.map((header) => {
-              const value = cleanData[header]
+              const value = syncedData[header]
               if (tsFields?.has(header) && typeof value === 'number') {
                 return unixToIso(value)
               }
@@ -998,11 +1002,11 @@ export function createDestination(
 
         const headers = (rows[0] as unknown[]).map((h) => String(h ?? ''))
         const idIdx = headers.indexOf('id')
-        const syncedAtIdx = headers.indexOf('_synced_at')
+        const syncedAtIdx = headers.indexOf('_last_synced_at')
         if (idIdx < 0 || syncedAtIdx < 0) {
           log.warn(
             { stream: streamName, headers },
-            'getStaleRecords: missing id or _synced_at column — skipping stream'
+            'getStaleRecords: missing id or _last_synced_at column — skipping stream'
           )
           continue
         }