Listen, Notify statements support for Postgresql

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added support for `LISTEN` and `NOTIFY` statements in PostgreSQL, including `bobgen-psql` parser support. Note: `LISTEN` requires a persistent connection and only registers the channel — receiving notifications must be implemented using your specific database driver. (thanks @manhrev)
+- Added `bobgen-psql` and `bobgen-sqlite` parser support for top-level `VALUES` queries. MySQL support is partial. (thanks @manhrev)
 - Generated `dberrors` packages now include generic and per-table check-constraint errors for PostgreSQL, matched by constraint name for `pq` and `pgx` drivers. (thanks @keithbro-imx)
 
 ### Changed

dialect/psql/dialect/listen.go

@@ -0,0 +1,20 @@
+package dialect
+
+import (
+	"context"
+	"io"
+
+	"github.com/stephenafamo/bob"
+)
+
+// Trying to represent the listen query structure as documented in
+// https://www.postgresql.org/docs/current/sql-listen.html
+type ListenQuery struct {
+	Channel string
+}
+
+func (l ListenQuery) WriteSQL(_ context.Context, w io.StringWriter, dl bob.Dialect, _ int) ([]any, error) {
+	w.WriteString("LISTEN ")
+	dl.WriteQuoted(w, l.Channel)
+	return nil, nil
+}

dialect/psql/dialect/notify.go

@@ -0,0 +1,26 @@
+package dialect
+
+import (
+	"context"
+	"io"
+
+	"github.com/stephenafamo/bob"
+)
+
+// Trying to represent the notify query structure as documented in
+// https://www.postgresql.org/docs/current/sql-notify.html
+type NotifyQuery struct {
+	Channel string
+	Payload string
+}
+
+func (n NotifyQuery) WriteSQL(_ context.Context, w io.StringWriter, dl bob.Dialect, _ int) ([]any, error) {
+	w.WriteString("NOTIFY ")
+	dl.WriteQuoted(w, n.Channel)
+	if n.Payload != "" {
+		w.WriteString(", '")
+		w.WriteString(n.Payload)
+		w.WriteString("'")
+	}
+	return nil, nil
+}

dialect/psql/listen.go

@@ -0,0 +1,19 @@
+package psql
+
+import (
+	"github.com/stephenafamo/bob"
+	"github.com/stephenafamo/bob/dialect/psql/dialect"
+)
+
+func Listen(mods ...bob.Mod[*dialect.ListenQuery]) bob.BaseQuery[*dialect.ListenQuery] {
+	q := &dialect.ListenQuery{}
+	for _, mod := range mods {
+		mod.Apply(q)
+	}
+
+	return bob.BaseQuery[*dialect.ListenQuery]{
+		Expression: q,
+		Dialect:    dialect.Dialect,
+		QueryType:  bob.QueryTypeListen,
+	}
+}

dialect/psql/listen_test.go

@@ -0,0 +1,21 @@
+package psql_test
+
+import (
+	"testing"
+
+	"github.com/stephenafamo/bob/dialect/psql"
+	"github.com/stephenafamo/bob/dialect/psql/lm"
+	testutils "github.com/stephenafamo/bob/test/utils"
+)
+
+func TestListen(t *testing.T) {
+	examples := testutils.Testcases{
+		"simple": {
+			Query:        psql.Listen(lm.Channel("my_channel")),
+			ExpectedSQL:  `LISTEN "my_channel"`,
+			ExpectedArgs: nil,
+		},
+	}
+
+	testutils.RunTests(t, examples, formatter)
+}

dialect/psql/lm/qm.go

@@ -0,0 +1,12 @@
+package lm
+
+import (
+	"github.com/stephenafamo/bob"
+	"github.com/stephenafamo/bob/dialect/psql/dialect"
+)
+
+func Channel(name string) bob.Mod[*dialect.ListenQuery] {
+	return bob.ModFunc[*dialect.ListenQuery](func(q *dialect.ListenQuery) {
+		q.Channel = name
+	})
+}

dialect/psql/nm/qm.go

@@ -0,0 +1,18 @@
+package nm
+
+import (
+	"github.com/stephenafamo/bob"
+	"github.com/stephenafamo/bob/dialect/psql/dialect"
+)
+
+func Channel(name string) bob.Mod[*dialect.NotifyQuery] {
+	return bob.ModFunc[*dialect.NotifyQuery](func(q *dialect.NotifyQuery) {
+		q.Channel = name
+	})
+}
+
+func Payload(p string) bob.Mod[*dialect.NotifyQuery] {
+	return bob.ModFunc[*dialect.NotifyQuery](func(q *dialect.NotifyQuery) {
+		q.Payload = p
+	})
+}

dialect/psql/notify.go

@@ -0,0 +1,19 @@
+package psql
+
+import (
+	"github.com/stephenafamo/bob"
+	"github.com/stephenafamo/bob/dialect/psql/dialect"
+)
+
+func Notify(mods ...bob.Mod[*dialect.NotifyQuery]) bob.BaseQuery[*dialect.NotifyQuery] {
+	q := &dialect.NotifyQuery{}
+	for _, mod := range mods {
+		mod.Apply(q)
+	}
+
+	return bob.BaseQuery[*dialect.NotifyQuery]{
+		Expression: q,
+		Dialect:    dialect.Dialect,
+		QueryType:  bob.QueryTypeNotify,
+	}
+}

dialect/psql/notify_test.go

@@ -0,0 +1,26 @@
+package psql_test
+
+import (
+	"testing"
+
+	"github.com/stephenafamo/bob/dialect/psql"
+	"github.com/stephenafamo/bob/dialect/psql/nm"
+	testutils "github.com/stephenafamo/bob/test/utils"
+)
+
+func TestNotify(t *testing.T) {
+	examples := testutils.Testcases{
+		"simple": {
+			Query:        psql.Notify(nm.Channel("my_channel")),
+			ExpectedSQL:  `NOTIFY "my_channel"`,
+			ExpectedArgs: nil,
+		},
+		"with payload": {
+			Query:        psql.Notify(nm.Channel("my_channel"), nm.Payload("hello world")),
+			ExpectedSQL:  `NOTIFY "my_channel", 'hello world'`,
+			ExpectedArgs: nil,
+		},
+	}
+
+	testutils.RunTests(t, examples, formatter)
+}

gen/bobgen-mysql/driver/parser/visitor.go

@@ -184,6 +184,8 @@ func (v *visitor) VisitSqlStatements(ctx *mysqlparser.SqlStatementsContext) any
 				v.Err = fmt.Errorf("stmt %d: could not get columns in select statement, got %T", i, resp)
 				return nil
 			}
+		case *mysqlparser.ValuesStatementContext:
+			queryType = bob.QueryTypeValues
 		}
 
 		allresp = append(allresp, StmtInfo{

gen/bobgen-psql/driver/parser/mods_listen.go

@@ -0,0 +1,11 @@
+package parser
+
+import (
+	"fmt"
+
+	pg "github.com/pganalyze/pg_query_go/v6"
+)
+
+func (w *walker) modListenStatement(stmt *pg.Node_ListenStmt, _ nodeInfo) {
+	fmt.Fprintf(w.mods, "q.Channel = %q\n", stmt.ListenStmt.Conditionname)
+}

gen/bobgen-psql/driver/parser/mods_notify.go

@@ -0,0 +1,14 @@
+package parser
+
+import (
+	"fmt"
+
+	pg "github.com/pganalyze/pg_query_go/v6"
+)
+
+func (w *walker) modNotifyStatement(stmt *pg.Node_NotifyStmt, _ nodeInfo) {
+	fmt.Fprintf(w.mods, "q.Channel = %q\n", stmt.NotifyStmt.Conditionname)
+	if stmt.NotifyStmt.Payload != "" {
+		fmt.Fprintf(w.mods, "q.Payload = %q\n", stmt.NotifyStmt.Payload)
+	}
+}

gen/bobgen-psql/driver/parser/mods_values.go

@@ -0,0 +1,49 @@
+package parser
+
+import (
+	"fmt"
+
+	pg "github.com/pganalyze/pg_query_go/v6"
+	"github.com/stephenafamo/bob/internal"
+)
+
+func (w *walker) modValuesStatement(stmt *pg.Node_SelectStmt, info nodeInfo) {
+	if orderInfo, ok := info.children["SortClause"]; ok {
+		w.editRules = append(w.editRules, internal.RecordPoints(
+			int(orderInfo.start),
+			int(orderInfo.end)-1,
+			func(start, end int) error {
+				fmt.Fprintf(w.mods, "q.AppendOrder(EXPR.subExpr(%d, %d))\n", start, end)
+				return nil
+			},
+		)...)
+	}
+
+	if limitInfo, ok := info.children["LimitCount"]; ok {
+		w.editRules = append(w.editRules, internal.RecordPoints(
+			int(limitInfo.start),
+			int(limitInfo.end)-1,
+			func(start, end int) error {
+				switch stmt.SelectStmt.LimitOption {
+				case pg.LimitOption_LIMIT_OPTION_COUNT:
+					fmt.Fprintf(w.mods, "q.SetLimit(EXPR.subExpr(%d, %d))\n", start, end)
+				case pg.LimitOption_LIMIT_OPTION_WITH_TIES:
+					w.imports = append(w.imports, []string{"github.com/stephenafamo/bob/clause"})
+					fmt.Fprintf(w.mods, "q.SetFetch(clause.Fetch{Count: EXPR.subExpr(%d, %d), WithTies: true})\n", start, end)
+				}
+				return nil
+			},
+		)...)
+	}
+
+	if offsetInfo, ok := info.children["LimitOffset"]; ok {
+		w.editRules = append(w.editRules, internal.RecordPoints(
+			int(offsetInfo.start),
+			int(offsetInfo.end)-1,
+			func(start, end int) error {
+				fmt.Fprintf(w.mods, "q.SetOffset(EXPR.subExpr(%d, %d))\n", start, end)
+				return nil
+			},
+		)...)
+	}
+}

gen/bobgen-psql/driver/parser/parser.go

@@ -89,6 +89,11 @@ func (p *Parser) ParseQuery(ctx context.Context, input string) (drivers.Query, e
 		return drivers.Query{}, fmt.Errorf("expected 1 statement, got %d", len(parseResult.Stmts))
 	}
 
+	stmt := parseResult.Stmts[0]
+	qType := getQueryType(stmt.Stmt)
+
+	var argTypes, resTypes []string
+
 	w := walker{
 		db:           p.db,
 		sharedSchema: p.sharedSchema,
@@ -104,17 +109,16 @@ func (p *Parser) ParseQuery(ctx context.Context, input string) (drivers.Query, e
 		paramIdxMap:  make(map[int64]int64),
 	}
 
-	stmt := parseResult.Stmts[0]
 	info := w.walk(stmt.Stmt)
 	switch node := stmt.Stmt.Node.(type) {
 	case *pg.Node_SelectStmt:
+		info = info.children["SelectStmt"]
 		if len(node.SelectStmt.ValuesLists) > 0 {
-			return drivers.Query{}, fmt.Errorf("VALUES statement is not supported")
+			w.modValuesStatement(node, info)
+		} else {
+			w.modSelectStatement(node, info)
 		}
 
-		info = info.children["SelectStmt"]
-		w.modSelectStatement(node, info)
-
 	case *pg.Node_InsertStmt:
 		info = info.children["InsertStmt"]
 		w.modInsertStatement(node, info)
@@ -130,6 +134,15 @@ func (p *Parser) ParseQuery(ctx context.Context, input string) (drivers.Query, e
 	case *pg.Node_MergeStmt:
 		info = info.children["MergeStmt"]
 		w.modMergeStatement(node, info)
+	case *pg.Node_ListenStmt:
+		// pg.ListenStmt has no Location field; find the keyword token directly
+		info = w.findTokenAfter(0, pg.Token_LISTEN)
+		w.modListenStatement(node, info)
+
+	case *pg.Node_NotifyStmt:
+		// pg.NotifyStmt has no Location field; find the keyword token directly
+		info = w.findTokenAfter(0, pg.Token_NOTIFY)
+		w.modNotifyStatement(node, info)
 	}
 
 	source := w.getSource(stmt.Stmt, info)
@@ -143,9 +156,12 @@ func (p *Parser) ParseQuery(ctx context.Context, input string) (drivers.Query, e
 		return drivers.Query{}, fmt.Errorf("format: %w", err)
 	}
 
-	argTypes, resTypes, err := p.getArgsAndCols(ctx, formatted)
-	if err != nil {
-		return drivers.Query{}, fmt.Errorf("get args and cols: %w", err)
+	// LISTEN/NOTIFY cannot be PREPAREd; they have no args or result columns
+	if qType != bob.QueryTypeListen && qType != bob.QueryTypeNotify {
+		argTypes, resTypes, err = p.getArgsAndCols(ctx, formatted)
+		if err != nil {
+			return drivers.Query{}, fmt.Errorf("get args and cols: %w", err)
+		}
 	}
 
 	if len(source.columns) != len(resTypes) {
@@ -223,6 +239,10 @@ func isReturningWithParseError(sql string, err error) bool {
 func getQueryType(stmt *pg.Node) bob.QueryType {
 	switch stmt.Node.(type) {
 	case *pg.Node_SelectStmt:
+		// VALUES (...) is parsed as SelectStmt with ValuesLists set; no separate node type exists
+		if len(stmt.Node.(*pg.Node_SelectStmt).SelectStmt.ValuesLists) > 0 {
+			return bob.QueryTypeValues
+		}
 		return bob.QueryTypeSelect
 	case *pg.Node_InsertStmt:
 		return bob.QueryTypeInsert
@@ -232,6 +252,10 @@ func getQueryType(stmt *pg.Node) bob.QueryType {
 		return bob.QueryTypeDelete
 	case *pg.Node_MergeStmt:
 		return bob.QueryTypeMerge
+	case *pg.Node_ListenStmt:
+		return bob.QueryTypeListen
+	case *pg.Node_NotifyStmt:
+		return bob.QueryTypeNotify
 	default:
 		return bob.QueryTypeUnknown
 	}