message-store: postgres: update getStream

add support for sql condition

message-store/postgres/get/stream.js

@@ -12,7 +12,7 @@ exports.createGetStream = (options) => {
     streamName,
     options.position,
     options.batchSize,
-    null
+    options.condition
   ]
 
   const parameters = '$1::varchar, $2::bigint, $3::bigint, $4::varchar'

message-store/postgres/test/message-store-postgres.test.js

@@ -49,6 +49,36 @@ describe('message-store-postgres', () => {
     })
   })
 
+  describe('getStream', () => {
+    describe('sql condition', () => {
+      describe('when not provided', () => {
+        it('parameter defaults to null, returns all results', async () => {
+          const log = createTestLog()
+          const messageStore = createMessageStore({ log })
+          const { streamName } = await examplePut(messageStore, { count: 3 })
+
+          const results = await messageStore.getStream(streamName)
+
+          expect(results.length).toBe(3)
+        })
+      })
+
+      describe('when provided', () => {
+        it('limits the results based on given sql condition', async () => {
+          const log = createTestLog()
+          const messageStore = createMessageStore({ log })
+          const { streamName, messages } = await examplePut(messageStore, { count: 3, trackMessages: true })
+
+          const CONDITION_SQL = 'messages.position = 1'
+          const results = await messageStore.getStream(streamName, { condition: CONDITION_SQL })
+
+          expect(results.length).toBe(1)
+          expect(results[0]).toMatchObject(messages[1])
+        })
+      })
+    })
+  })
+
   describe('get-last', () => {
     describe('connection error', () => {
       it('propagates error', async () => {

test/postgres-automated-test-schema.sql

@@ -1,3 +1,9 @@
+-- configurations
+
+ALTER DATABASE message_store SET message_store.sql_condition TO on; 
+
+-- automated_tests
+
 CREATE TABLE IF NOT EXISTS message_store.automated_tests (
   id UUID NOT NULL,
   time TIMESTAMP WITHOUT TIME ZONE DEFAULT (now() AT TIME ZONE 'utc') NOT NULL