change soft to hard delete

.vscode/launch.json

@@ -0,0 +1,18 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Attach to serve (tsx)",
+            "type": "node",
+            "request": "attach",
+            "port": 9229,
+            "restart": true,
+            "skipFiles": [
+                "<node_internals>/**"
+            ]
+        }
+    ]
+}

apps/engine/src/lib/pipeline.test.ts

@@ -156,6 +156,52 @@ describe('enforceCatalog()', () => {
     })
   })
 
+  it('strips the `deleted` field from data even when no json_schema is configured', async () => {
+    const msgs: Message[] = [
+      {
+        type: 'record',
+        record: {
+          stream: 'customers',
+          recordDeleted: true,
+          data: { id: 'cus_1', name: 'Alice', deleted: true },
+          emitted_at: '2024-01-01T00:00:00.000Z',
+        },
+      },
+    ]
+    const result = await drain(enforceCatalog(catalog([{ name: 'customers' }]))(toAsync(msgs)))
+    expect(result).toHaveLength(1)
+    expect((result[0] as any).record.data).toEqual({ id: 'cus_1', name: 'Alice' })
+    expect((result[0] as any).record.recordDeleted).toBe(true)
+  })
+
+  it('strips the `deleted` field from data even when json_schema declares it', async () => {
+    const msgs: Message[] = [
+      {
+        type: 'record',
+        record: {
+          stream: 'customers',
+          data: { id: 'cus_1', deleted: false },
+          emitted_at: '2024-01-01T00:00:00.000Z',
+        },
+      },
+    ]
+    const result = await drain(
+      enforceCatalog(
+        catalog([
+          {
+            name: 'customers',
+            json_schema: {
+              type: 'object',
+              properties: { id: { type: 'string' }, deleted: { type: 'boolean' } },
+            },
+          },
+        ])
+      )(toAsync(msgs))
+    )
+    expect(result).toHaveLength(1)
+    expect((result[0] as any).record.data).toEqual({ id: 'cus_1' })
+  })
+
   it('drops record with unknown stream and logs error', async () => {
     const msgs: Message[] = [
       {

apps/engine/src/lib/pipeline.ts

@@ -9,6 +9,9 @@ import { log } from '../logger.js'
 
 // MARK: - enforceCatalog
 
+/** Fields that destinations never persist; tombstoning is signalled via `recordDeleted`. */
+const STRIPPED_RECORD_FIELDS = new Set(['deleted'])
+
 /**
  * Drop messages for streams not in the catalog and apply per-stream field filtering.
  * Passes non-data messages (log, trace, catalog) through unchanged.
@@ -26,19 +29,13 @@ export function enforceCatalog<T extends Message>(
           continue
         }
         const props = cs.stream.json_schema?.properties as Record<string, unknown> | undefined
-        if (props) {
-          yield {
-            ...msg,
-            record: {
-              ...msg.record,
-              data: Object.fromEntries(
-                Object.entries(msg.record.data).filter(([key]) => key in props)
-              ),
-            },
-          }
-        } else {
-          yield msg
+        const filtered: Record<string, unknown> = {}
+        for (const [key, value] of Object.entries(msg.record.data)) {
+          if (STRIPPED_RECORD_FIELDS.has(key)) continue
+          if (props && !(key in props)) continue
+          filtered[key] = value
         }
+        yield { ...msg, record: { ...msg.record, data: filtered } }
       } else if (msg.type === 'source_state') {
         if (msg.source_state.state_type === 'global') {
           yield msg // global state needs no catalog validation