Work not only in the public schema (#76)

Author: vgarvardt

.github/workflows/testing.yml

@@ -1,3 +1,4 @@
+---
 name: Testing
 
 on:
@@ -12,9 +13,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v2
+        with:
+          only-new-issues: ${{ github.event_name == 'pull_request' }}
 
   go-test:
     name: Go ${{ matrix.go }} Test
@@ -23,7 +26,7 @@ jobs:
 
     strategy:
       matrix:
-        go: [ '~1.17', '~1.18' ]
+        go: [ '1.17', '1.18', '1.19' ]
 
     services:
       postgres:
@@ -41,12 +44,13 @@ jobs:
           --health-retries 5
 
     steps:
+      - name: Check out code
+        uses: actions/checkout@v3
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v3
         with:
           go-version: ${{ matrix.go }}
-      - name: Check out code
-        uses: actions/checkout@v2
+
       - name: Run unit tests
         run: make test-unit
       - name: Run examples tests
@@ -64,3 +68,41 @@ jobs:
     steps:
       - name: Done
         run: echo "All Golang versions tests are Green!"
+
+  test-non-public-schema:
+    name: Test non-public PG schema
+    runs-on: ubuntu-latest
+    needs: [ lint ]
+
+    services:
+      postgres:
+        image: postgres:10
+        ports:
+          - "5432"
+        env:
+          POSTGRES_USER: goengine
+          POSTGRES_PASSWORD: goengine
+          POSTGRES_DB: goengine
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version-file: ./go.mod
+
+      - name: Run unit tests
+        run: make test-unit
+      - name: Run examples tests
+        run: make test-examples
+      - name: Run integration tests
+        run: make test-integration
+        env:
+          # integration tests suite creates new DB per test run and initialises schema as well
+          POSTGRES_DSN: 'postgres://goengine:goengine@localhost:${{ job.services.postgres.ports[5432] }}/goengine?sslmode=disable&search_path=custom'

driver/sql/postgres/eventstore.go

@@ -281,14 +281,21 @@ func (e *EventStore) tableName(s goengine.StreamName) (string, error) {
 }
 
 func (e *EventStore) tableExists(ctx context.Context, tableName string) bool {
+	var currentSchema string
+	if err := e.db.QueryRowContext(ctx, `select current_schema()`).Scan(&currentSchema); err != nil {
+		e.logger.Warn("error on getting current schema", func(e goengine.LoggerEntry) {
+			e.Error(err)
+		})
+
+		return false
+	}
+
 	var exists bool
-	err := e.db.QueryRowContext(
+	if err := e.db.QueryRowContext(
 		ctx,
-		`SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1)`,
-		tableName,
-	).Scan(&exists)
-
-	if err != nil {
+		`SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)`,
+		currentSchema, tableName,
+	).Scan(&exists); err != nil {
 		e.logger.Warn("error on reading from information_schema", func(e goengine.LoggerEntry) {
 			e.Error(err)
 			e.String("table", tableName)

driver/sql/postgres/eventstore_test.go

@@ -11,8 +11,11 @@ import (
 	"testing"
 	"time"
 
-	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	"github.com/hellofresh/goengine/v2"
 	"github.com/hellofresh/goengine/v2/aggregate"
 	driverSQL "github.com/hellofresh/goengine/v2/driver/sql"
@@ -22,8 +25,6 @@ import (
 	"github.com/hellofresh/goengine/v2/mocks"
 	mockSQL "github.com/hellofresh/goengine/v2/mocks/driver/sql"
 	strategyPostgres "github.com/hellofresh/goengine/v2/strategy/json/sql/postgres"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
 func TestNewEventStore(t *testing.T) {
@@ -384,8 +385,11 @@ func TestEventStore_Load(t *testing.T) {
 }
 
 func mockHasStreamQuery(result bool, mock sqlmock.Sqlmock) {
-	mockRows := sqlmock.NewRows([]string{"type"}).AddRow(result)
-	mock.ExpectQuery(`SELECT EXISTS\((.+)`).WithArgs("events_orders").WillReturnRows(mockRows)
+	schemaRows := sqlmock.NewRows([]string{"current_schema"}).AddRow("public")
+	mock.ExpectQuery(`select current_schema()`).WillReturnRows(schemaRows)
+
+	existsRows := sqlmock.NewRows([]string{"type"}).AddRow(result)
+	mock.ExpectQuery(`SELECT EXISTS\((.+)`).WithArgs("public", "events_orders").WillReturnRows(existsRows)
 }
 
 func mockMessages(ctrl *gomock.Controller) (*mocks.MessagePayloadConverter, []goengine.Message) {

extension/amqp/amqp.go

@@ -3,8 +3,9 @@ package amqp
 import (
 	"io"
 
+	amqp "github.com/rabbitmq/amqp091-go"
+
 	"github.com/hellofresh/goengine/v2"
-	"github.com/streadway/amqp"
 )
 
 // NotificationChannel represents a channel for notifications

extension/amqp/amqp_test.go

@@ -6,10 +6,11 @@ package amqp_test
 import (
 	"testing"
 
+	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/hellofresh/goengine/v2"
 	goengineAmqp "github.com/hellofresh/goengine/v2/extension/amqp"
-	"github.com/streadway/amqp"
-	"github.com/stretchr/testify/assert"
 )
 
 type mockAcknowledger struct {

extension/amqp/listener.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	"github.com/mailru/easyjson"
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 
 	"github.com/hellofresh/goengine/v2"
 	"github.com/hellofresh/goengine/v2/driver/sql"

extension/amqp/listener_test.go

@@ -10,9 +10,9 @@ import (
 	"testing"
 	"time"
 
+	libamqp "github.com/rabbitmq/amqp091-go"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
-	libamqp "github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 

extension/amqp/publisher.go

@@ -6,10 +6,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/mailru/easyjson"
+	amqp "github.com/rabbitmq/amqp091-go"
+
 	"github.com/hellofresh/goengine/v2"
 	"github.com/hellofresh/goengine/v2/driver/sql"
-	"github.com/mailru/easyjson"
-	"github.com/streadway/amqp"
 )
 
 var _ sql.ProjectionTrigger = (&NotificationPublisher{}).Publish

go.mod

@@ -9,8 +9,8 @@ require (
 	github.com/lib/pq v1.10.7
 	github.com/mailru/easyjson v0.7.7
 	github.com/prometheus/client_golang v1.13.0
+	github.com/rabbitmq/amqp091-go v1.5.0
 	github.com/sirupsen/logrus v1.9.0
-	github.com/streadway/amqp v1.0.0
 	github.com/stretchr/testify v1.8.0
 	go.uber.org/zap v1.23.0
 )
@@ -31,7 +31,6 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/sys v0.0.0-20220909162455-aba9fc2a8ff2 // indirect
 	golang.org/x/tools v0.1.12 // indirect
-	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )