Removed redundant stream position from streams tables

Author: oskardudycz

src/packages/emmett-postgresql/src/eventStore/schema/appendToStream.ts

@@ -48,9 +48,15 @@ export const appendToStreamSQL = rawSql(
       v_transaction_id := pg_current_xact_id();
 
       IF v_expected_stream_position IS NULL THEN
-          SELECT COALESCE(stream_position, 0) INTO v_expected_stream_position
-          FROM ${streamsTable.name}
-          WHERE stream_id = v_stream_id AND partition = v_partition AND is_archived = FALSE;
+          SELECT COALESCE(
+            (SELECT stream_position 
+            FROM ${streamsTable.name}
+            WHERE stream_id = v_stream_id 
+              AND partition = v_partition 
+              AND is_archived = FALSE
+            LIMIT 1), 
+            0
+        ) INTO v_expected_stream_position;
       END IF;
 
       v_next_stream_position := v_expected_stream_position + array_upper(v_messages_data, 1);

src/packages/emmett-postgresql/src/eventStore/schema/tables.ts

@@ -15,9 +15,12 @@ export const streamsTableSQL = rawSql(
       stream_type       TEXT                      NOT NULL,
       stream_metadata   JSONB                     NOT NULL,
       is_archived       BOOLEAN                   NOT NULL DEFAULT FALSE,
-      PRIMARY KEY (stream_id, stream_position, partition, is_archived),
-      UNIQUE (stream_id, partition, is_archived)
-  ) PARTITION BY LIST (partition);`,
+      PRIMARY KEY (stream_id, partition, is_archived)
+  ) PARTITION BY LIST (partition);
+   
+  CREATE UNIQUE INDEX idx_streams_unique 
+  ON ${streamsTable.name}(stream_id, partition, is_archived) 
+  INCLUDE (stream_position);`,
 );
 
 export const messagesTableSQL = rawSql(

src/packages/emmett-sqlite/src/eventStore/schema/tables.ts

@@ -16,7 +16,7 @@ export const streamsTableSQL = sql(
       stream_type       TEXT                      NOT NULL,
       stream_metadata   JSONB                     NOT NULL,
       is_archived       BOOLEAN                   NOT NULL DEFAULT FALSE,
-      PRIMARY KEY (stream_id, stream_position, partition, is_archived),
+      PRIMARY KEY (stream_id, partition, is_archived),
       UNIQUE (stream_id, partition, is_archived)
   );`,
 );