From c671f73ee5e63cf43033254e5116b8d66dd7d55d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aslak=20Helles=C3=B8y?= <aslak.hellesoy@gmail.com>
Date: Thu, 7 Dec 2023 21:45:56 +0000
Subject: [PATCH 1/2] Merge streamed updates with existing record

---
 CHANGELOG.md                                  |  3 ++
 .../merges-updates-with-previous-value.md     | 24 ++++++++++++++
 src/index.ts                                  | 11 +++++--
 test/liveTableBuffering.test.ts               | 31 +++++++++++++++++++
 4 files changed, 66 insertions(+), 3 deletions(-)
 create mode 100644 docs/sequence-diagrams/merges-updates-with-previous-value.md

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e3a2cdd..2fef841 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Fixed
+- Merge streamed updates with existing record. Fixes [#2](https://github.com/openartmarket/supabase-live-table/issues/2)
+
 ## [0.0.7] - 2023-09-23
 ### Fixed
 - Ignore updates that arrive after the snapshot, but are timestamped before the snapshot
diff --git a/docs/sequence-diagrams/merges-updates-with-previous-value.md b/docs/sequence-diagrams/merges-updates-with-previous-value.md
new file mode 100644
index 0000000..9cc8a55
--- /dev/null
+++ b/docs/sequence-diagrams/merges-updates-with-previous-value.md
@@ -0,0 +1,24 @@
+### merges updates with previous value
+
+```mermaid
+sequenceDiagram
+  LiveTable->>+Supabase: subscribe
+  Supabase->>-LiveTable: subscription active
+  LiveTable->>+Supabase: get snapshot
+  Supabase-->>LiveTable: UPDATE {"id":1,"created_at":"1","updated_at":"3","type":"vehicle","color":"red"}
+  Supabase->>-LiveTable: snaphot: [{"id":1,"created_at":"1","updated_at":"2","type":"vehicle","name":"Bicycle","color":"black"}]
+```
+
+### replica
+```json
+[
+  {
+    "id": 1,
+    "created_at": "1",
+    "updated_at": "3",
+    "type": "vehicle",
+    "name": "Bicycle",
+    "color": "red"
+  }
+]
+```
diff --git a/src/index.ts b/src/index.ts
index 6cfe61a..3c7e391 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -113,7 +113,7 @@ type Insert<TableRow extends LiveRow> = {
 
 type Update<TableRow extends LiveRow> = {
   type: 'UPDATE';
-  record: TableRow;
+  record: Partial<TableRow>;
   timestamp: string;
 };
 
@@ -180,10 +180,15 @@ export class LiveTable<TableRow extends LiveRow> implements ILiveTable<TableRow>
         break;
       }
       case 'UPDATE': {
-        if (!this.recordById.has(record.id)) {
+        const id = record.id;
+        if (!id) {
+          throw new Error(`Cannot delete. Record has no id: ${JSON.stringify(record)}`);
+        }
+        const oldRecord = this.recordById.get(id);
+        if (oldRecord === undefined) {
           throw new Error(`Cannot update. Record does not exist: ${JSON.stringify(record)}`);
         }
-        this.recordById.set(record.id, record);
+        this.recordById.set(id, { ...oldRecord, ...record });
         break;
       }
       case 'DELETE': {
diff --git a/test/liveTableBuffering.test.ts b/test/liveTableBuffering.test.ts
index 4b03421..c177c31 100644
--- a/test/liveTableBuffering.test.ts
+++ b/test/liveTableBuffering.test.ts
@@ -178,6 +178,37 @@ describe('LiveTable Buffering', () => {
     await lt.close();
   });
 
+  it('merges updates with previous value', async (test) => {
+    const lt = new MermaidLiveTable(new LiveTable<ThingRow>(parseTimestamp), test.task.name);
+
+    lt.subscribe();
+    lt.subscribed();
+    lt.requestSnapshot();
+
+    const streamRecord: Partial<ThingRow> = {
+      id: 1,
+      created_at: t1,
+      updated_at: t3,
+      type: 'vehicle',
+      color: 'red',
+    };
+    lt.processEvent({ timestamp: t3, type: 'UPDATE', record: streamRecord });
+
+    const snapshotRecord = {
+      id: 1,
+      created_at: t1,
+      updated_at: t2,
+      type: 'vehicle',
+      name: 'Bicycle',
+      color: 'black',
+    };
+    lt.processSnapshot([snapshotRecord]);
+
+    expect(lt.records).toEqual([{ ...snapshotRecord, color: 'red', updated_at: t3 }]);
+
+    await lt.close();
+  });
+
   it('replays deletes that arrived after the snapshot', async (test) => {
     const lt = new MermaidLiveTable(new LiveTable<ThingRow>(parseTimestamp), test.task.name);
 

From 46110a831574fe6a0f0fc291af80f8bf064282a0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aslak=20Helles=C3=B8y?= <aslak.hellesoy@gmail.com>
Date: Thu, 7 Dec 2023 22:06:17 +0000
Subject: [PATCH 2/2] Fix typo

---
 src/index.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/index.ts b/src/index.ts
index 3c7e391..30e34ab 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -182,7 +182,7 @@ export class LiveTable<TableRow extends LiveRow> implements ILiveTable<TableRow>
       case 'UPDATE': {
         const id = record.id;
         if (!id) {
-          throw new Error(`Cannot delete. Record has no id: ${JSON.stringify(record)}`);
+          throw new Error(`Cannot update. Record has no id: ${JSON.stringify(record)}`);
         }
         const oldRecord = this.recordById.get(id);
         if (oldRecord === undefined) {