From 342aad6fe2eb5b9cd5fd1bb05e7dfb0e15f0a039 Mon Sep 17 00:00:00 2001
From: Adriano Caloiaro <code@adriano.fyi>
Date: Tue, 29 Aug 2023 11:09:47 +0200
Subject: [PATCH] fix: Incorrectly passing pgx connections params to pq

Packages `pq` and `pgx` allow different connection string parameters.
Since `pgx` is neoq's primary SQL library, but migrations use `pq`,
we had to ensure that pgx connection parameters do not get passed to
the pq library when running migrations.

This issue was introduced in version `0.20.0`.
---
 .goreleaser.yaml                      |  8 +++++++
 backends/postgres/postgres_backend.go | 33 +++++++++++++++++++++++----
 2 files changed, 36 insertions(+), 5 deletions(-)

diff --git a/.goreleaser.yaml b/.goreleaser.yaml
index 5a17771..d3f598c 100644
--- a/.goreleaser.yaml
+++ b/.goreleaser.yaml
@@ -31,6 +31,14 @@ changelog:
     exclude:
       - '^docs:'
       - '^test:'
+  groups:
+  - title: Features
+    regexp: '^.*?feat(\([[:word:]]+\))??!?:.+$'
+    order: 0
+  - title: "Bug fixes"
+    regexp: '^.*?fix(\([[:word:]]+\))??!?:.+$'
+  - title: Others
+    order: 999
 
 # The lines beneath this are called `modelines`. See `:help modeline`
 # Feel free to remove those if you don't want/use them.
diff --git a/backends/postgres/postgres_backend.go b/backends/postgres/postgres_backend.go
index da07719..67c290e 100644
--- a/backends/postgres/postgres_backend.go
+++ b/backends/postgres/postgres_backend.go
@@ -126,7 +126,7 @@ func Backend(ctx context.Context, opts ...config.Option) (pb types.Backend, err
 	p.cancelFuncs = append(p.cancelFuncs, cancel)
 	p.mu.Unlock()
 
-	err = p.initializeDB(p.config.ConnectionString)
+	err = p.initializeDB()
 	if err != nil {
 		return
 	}
@@ -197,14 +197,37 @@ func txFromContext(ctx context.Context) (t pgx.Tx, err error) {
 // initializeDB initializes the tables, types, and indices necessary to operate Neoq
 //
 //nolint:funlen,gocyclo,cyclop
-func (p *PgBackend) initializeDB(connStr string) (err error) {
+func (p *PgBackend) initializeDB() (err error) {
 	d, err := iofs.New(migrationsFS, "migrations")
 	if err != nil {
 		p.logger.Error("unable to run migrations", "error", err)
 		return
 	}
 
-	m, err := migrate.NewWithSourceInstance("iofs", d, connStr)
+	// `pgx` supports config params that `pq` does not. Since pgx is neoq's primary SQL interface, user often configure
+	// it with pgx-specific config params like `max_conn_count`. However, `go-migrate` uses `pq` under the hood, and
+	// these `pgx` config params cause `pq` to throw an "unknown config parameter" error when they're encountered.
+	// So we must first sanitize connection strings for pq
+	var pgxCfg *pgx.ConnConfig
+	pgxCfg, err = pgx.ParseConfig(p.config.ConnectionString)
+	if err != nil {
+		p.logger.Error("unable to run migrations", "error", err)
+		return
+	}
+
+	sslMode := "verify-ca"
+	// nil TLSConfig means "sslmode=disable" was set on the connection
+	if pgxCfg.TLSConfig == nil {
+		sslMode = "disable"
+	}
+
+	pqConnectionString := fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=%s",
+		pgxCfg.User,
+		pgxCfg.Password,
+		pgxCfg.Host,
+		pgxCfg.Database,
+		sslMode)
+	m, err := migrate.NewWithSourceInstance("iofs", d, pqConnectionString)
 	if err != nil {
 		p.logger.Error("unable to run migrations", "error", err)
 		return
@@ -282,9 +305,9 @@ func (p *PgBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, e
 		}
 	} else {
 		p.mu.Lock()
-		p.logger.Debug("added job to future jobs list", "job_id", jobID, "run_after", job.RunAfter)
 		p.futureJobs[jobID] = job.RunAfter
 		p.mu.Unlock()
+		p.logger.Debug("added job to future jobs list", "job_id", jobID, "run_after", job.RunAfter)
 	}
 
 	err = tx.Commit(ctx)
@@ -731,7 +754,7 @@ func (p *PgBackend) handleJob(ctx context.Context, jobID string, h handler.Handl
 // This will lead to jobs not getting processed until the worker is restarted.
 // Implement disconnect handling.
 func (p *PgBackend) listen(ctx context.Context, queue string) (c chan string, ready chan bool) {
-	c = make(chan string)
+	c = make(chan string, p.handlers[queue].Concurrency)
 	ready = make(chan bool)
 
 	go func(ctx context.Context) {