feat: add LISTEN/NOTIFY and commit LSN support

pkg/analyzer/analyzer.go

@@ -95,6 +95,10 @@ func (a *Analyzer) Analyze(stmt parser.Statement) error {
 	case *parser.CreateIndexStmt, *parser.DropIndexStmt:
 		// Index statements don't need semantic analysis
 		return nil
+	case *parser.ListenStmt, *parser.UnlistenStmt, *parser.NotifyStmt:
+		// Channel names and payloads are validated by the parser; there is no
+		// schema or expression to analyze.
+		return nil
 	default:
 		return &AnalysisError{
 			Type:    ErrUnknown,

pkg/capabilities/capabilities.go

@@ -36,6 +36,7 @@ type Document struct {
 	GeneratedAt   string `json:"generated_at,omitempty"`
 
 	Protocol      Protocol          `json:"protocol"`
+	Notifications Notifications     `json:"notifications"`
 	Constraints   Constraints       `json:"constraints"`
 	DDL           DDL               `json:"ddl"`
 	Transactions  Transactions      `json:"transactions"`
@@ -54,6 +55,17 @@ type Protocol struct {
 	TLS            string `json:"tls"`
 }
 
+type Notifications struct {
+	Listen         Feature `json:"listen"`
+	Unlisten       Feature `json:"unlisten"`
+	Notify         Feature `json:"notify"`
+	PayloadLimit   int     `json:"payload_limit_bytes"`
+	ChannelLimit   int     `json:"channel_limit_bytes"`
+	Transactional  bool    `json:"transactional"`
+	DurableReplay  bool    `json:"durable_replay"`
+	CommitLSNParam string  `json:"commit_lsn_parameter"`
+}
+
 type Constraints struct {
 	NotNull       Feature `json:"not_null"`
 	Unique        Feature `json:"unique"`
@@ -133,6 +145,17 @@ func Current() *Document {
 			TLS:            "terminated by the managed pgproxy, not the engine listener",
 		},
 
+		Notifications: Notifications{
+			Listen:         Feature{Supported: true, Status: "enforced"},
+			Unlisten:       Feature{Supported: true, Status: "enforced", Notes: "UNLISTEN * removes every subscription"},
+			Notify:         Feature{Supported: true, Status: "enforced", Notes: "duplicate (channel,payload) pairs collapse within one transaction"},
+			PayloadLimit:   8000,
+			ChannelLimit:   63,
+			Transactional:  true,
+			DurableReplay:  false,
+			CommitLSNParam: "pizzasql.lsn",
+		},
+
 		Constraints: Constraints{
 			NotNull:       Feature{Supported: true, Status: "enforced"},
 			Unique:        Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "unique indexes, INLINE UNIQUE, and ON CONFLICT are enforced atomically"},

pkg/capabilities/capabilities_test.go

@@ -33,4 +33,10 @@ func TestCurrentDocumentIsVersionedAndMarshalable(t *testing.T) {
 	if round.ErrorCodes["0A000"] != "feature_not_supported" {
 		t.Fatalf("0A000 mapping = %q", round.ErrorCodes["0A000"])
 	}
+	if !round.Notifications.Listen.Supported || !round.Notifications.Notify.Supported {
+		t.Fatal("LISTEN/NOTIFY should be advertised as supported")
+	}
+	if round.Notifications.PayloadLimit != 8000 || round.Notifications.DurableReplay {
+		t.Fatalf("unexpected notification limits: %+v", round.Notifications)
+	}
 }

pkg/executor/executor.go

@@ -42,6 +42,17 @@ type Executor struct {
 	savepoints         []string // stack of savepoint names
 	savepointPositions []int    // session mutation-log positions for each savepoint
 
+	// LISTEN/NOTIFY state. The hub and subscriber are per database instance;
+	// listenOps and notifyQueue are staged until COMMIT so LISTEN and NOTIFY
+	// observe transactional visibility.
+	hub               *storage.Hub
+	subscriber        *storage.Subscriber
+	backendPID        int32
+	listens           map[string]bool
+	listenOps         []listenOp
+	notifyQueue       []storage.Notification
+	notificationSaves []notificationSnapshot
+
 	// Subquery context for correlated subqueries
 	outerRow storage.Row
 
@@ -99,6 +110,7 @@ func New(schema *storage.SchemaManager, table *storage.TableManager) *Executor {
 		attachedDatabases: make(map[string]*DatabaseConnection),
 		currentDatabase:   "main",
 		views:             make(map[string]*parser.SelectStmt),
+		listens:           make(map[string]bool),
 	}
 
 	// Expression indexes persist only SQL text, so the storage layer needs a
@@ -213,6 +225,12 @@ func (e *Executor) Execute(stmt parser.Statement) (*Result, error) {
 		return e.executeDetach(s)
 	case *parser.AnalyzeStmt:
 		return e.executeAnalyze(s)
+	case *parser.ListenStmt:
+		return e.executeListen(s)
+	case *parser.UnlistenStmt:
+		return e.executeUnlisten(s)
+	case *parser.NotifyStmt:
+		return e.executeNotify(s)
 	}
 
 	// SQLite catalog/metadata dispatch. SELECTs against sqlite_master /
@@ -4083,6 +4101,7 @@ func (e *Executor) executeBegin(stmt *parser.BeginStmt) (*Result, error) {
 	e.inTransaction = true
 	e.savepoints = nil
 	e.savepointPositions = nil
+	e.clearNotificationState()
 
 	result := NewResult("BEGIN")
 	return result, nil
@@ -4094,18 +4113,22 @@ func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) {
 		return nil, fmt.Errorf("cannot commit: no transaction in progress")
 	}
 
-	if err := e.session.Commit(); err != nil {
+	lsn, err := e.session.CommitWithLSN()
+	if err != nil {
 		e.inTransaction = false
 		e.savepoints = nil
 		e.savepointPositions = nil
+		e.finishRollback()
 		return nil, err
 	}
 
 	e.inTransaction = false
 	e.savepoints = nil
 	e.savepointPositions = nil
+	e.finishCommit(lsn)
 
 	result := NewResult("COMMIT")
+	result.LSN = lsn
 	return result, nil
 }
 
@@ -4126,6 +4149,7 @@ func (e *Executor) executeRollback(stmt *parser.RollbackStmt) (*Result, error) {
 	e.inTransaction = false
 	e.savepoints = nil
 	e.savepointPositions = nil
+	e.finishRollback()
 	// total_changes() is monotonic (SQLite semantics): it is NOT decremented on
 	// rollback. last_insert_rowid() and changes() also intentionally hold.
 
@@ -4143,11 +4167,13 @@ func (e *Executor) executeSavepoint(stmt *parser.SavepointStmt) (*Result, error)
 		e.inTransaction = true
 		e.savepoints = nil
 		e.savepointPositions = nil
+		e.clearNotificationState()
 	}
 
 	// Add savepoint marker
 	e.savepoints = append(e.savepoints, stmt.Name)
 	e.savepointPositions = append(e.savepointPositions, e.session.Snapshot())
+	e.notificationSaves = append(e.notificationSaves, e.notificationSnapshot())
 
 	result := NewResult("SAVEPOINT")
 	return result, nil
@@ -4165,6 +4191,9 @@ func (e *Executor) executeRelease(stmt *parser.ReleaseStmt) (*Result, error) {
 		if e.savepoints[i] == stmt.Name {
 			e.savepoints = e.savepoints[:i]
 			e.savepointPositions = e.savepointPositions[:i]
+			if i <= len(e.notificationSaves) {
+				e.notificationSaves = e.notificationSaves[:i]
+			}
 			found = true
 			break
 		}
@@ -4262,6 +4291,10 @@ func (e *Executor) rollbackToSavepoint(name string) (*Result, error) {
 
 	logPosition := e.savepointPositions[savepointIdx]
 	e.session.RollbackTo(logPosition)
+	if savepointIdx < len(e.notificationSaves) {
+		e.rollbackNotificationsTo(e.notificationSaves[savepointIdx])
+		e.notificationSaves = e.notificationSaves[:savepointIdx+1]
+	}
 
 	// Remove savepoints after the target
 	e.savepoints = e.savepoints[:savepointIdx+1]

pkg/executor/notify.go

@@ -0,0 +1,204 @@
+package executor
+
+import (
+	"fmt"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// listenOp records one LISTEN/UNLISTEN statement so its effect can be applied
+// at transaction commit. PostgreSQL makes subscription changes visible only
+// after commit; a rollback leaves the previous subscription set untouched.
+type listenOp struct {
+	kind    listenOpKind
+	channel string
+}
+
+type listenOpKind int
+
+const (
+	listenOpListen listenOpKind = iota
+	listenOpUnlisten
+	listenOpUnlistenAll
+)
+
+// notificationSnapshot captures the staged notification state at a savepoint.
+type notificationSnapshot struct {
+	listenOps   int
+	notifyQueue int
+}
+
+// SetNotificationHub installs the per-database notification registry and
+// creates this session's subscriber. It must be called before the executor
+// processes LISTEN or NOTIFY. Safe to call once per connection.
+func (e *Executor) SetNotificationHub(hub *storage.Hub) {
+	e.hub = hub
+	if hub == nil {
+		return
+	}
+	if e.subscriber == nil {
+		e.subscriber = storage.NewSubscriber(e.backendPID, 0)
+	}
+}
+
+// SetBackendPID sets the backend PID reported in NotificationResponse frames.
+// It also applies to an already-created subscriber.
+func (e *Executor) SetBackendPID(pid int32) {
+	e.backendPID = pid
+	if e.subscriber == nil && e.hub != nil {
+		e.subscriber = storage.NewSubscriber(pid, 0)
+	}
+}
+
+// NotificationChannel returns the receive side of this session's notification
+// queue, or nil when no hub is configured.
+func (e *Executor) NotificationChannel() <-chan storage.Notification {
+	if e.subscriber == nil {
+		return nil
+	}
+	return e.subscriber.Notifications()
+}
+
+// NotificationOverflow is closed when the session's outbound queue is full and
+// the hub has dropped it. Protocol layers should disconnect the connection.
+func (e *Executor) NotificationOverflow() <-chan struct{} {
+	if e.subscriber == nil {
+		return nil
+	}
+	return e.subscriber.Overflow()
+}
+
+// Close releases this session's subscriptions. It is the disconnect path; it
+// is safe to call more than once.
+func (e *Executor) Close() error {
+	if e.hub != nil && e.subscriber != nil {
+		e.hub.UnsubscribeAll(e.subscriber)
+	}
+	e.listens = make(map[string]bool)
+	e.listenOps = nil
+	e.notifyQueue = nil
+	e.notificationSaves = nil
+	return e.RollbackActive()
+}
+
+func (e *Executor) executeListen(stmt *parser.ListenStmt) (*Result, error) {
+	if stmt.Channel == "" {
+		return nil, fmt.Errorf("channel name cannot be empty")
+	}
+	if e.inTransaction {
+		e.listenOps = append(e.listenOps, listenOp{kind: listenOpListen, channel: stmt.Channel})
+		return NewResult("LISTEN"), nil
+	}
+	e.applyListenOp(listenOp{kind: listenOpListen, channel: stmt.Channel})
+	return NewResult("LISTEN"), nil
+}
+
+func (e *Executor) executeUnlisten(stmt *parser.UnlistenStmt) (*Result, error) {
+	op := listenOp{kind: listenOpUnlisten, channel: stmt.Channel}
+	if stmt.All {
+		op = listenOp{kind: listenOpUnlistenAll}
+	} else if stmt.Channel == "" {
+		return nil, fmt.Errorf("channel name cannot be empty")
+	}
+	if e.inTransaction {
+		e.listenOps = append(e.listenOps, op)
+		return NewResult("UNLISTEN"), nil
+	}
+	e.applyListenOp(op)
+	return NewResult("UNLISTEN"), nil
+}
+
+func (e *Executor) executeNotify(stmt *parser.NotifyStmt) (*Result, error) {
+	if len(stmt.Payload) > 8000 {
+		return nil, fmt.Errorf("notification payload is longer than 8000 bytes")
+	}
+	n := storage.Notification{
+		PID:     e.backendPID,
+		Channel: stmt.Channel,
+		Payload: stmt.Payload,
+	}
+	if !e.inTransaction {
+		e.publish(n)
+		return NewResult("NOTIFY"), nil
+	}
+	// PostgreSQL collapses duplicate (channel, payload) pairs within one
+	// transaction, regardless of how many times they were raised.
+	for _, queued := range e.notifyQueue {
+		if queued.Channel == n.Channel && queued.Payload == n.Payload {
+			return NewResult("NOTIFY"), nil
+		}
+	}
+	e.notifyQueue = append(e.notifyQueue, n)
+	return NewResult("NOTIFY"), nil
+}
+
+func (e *Executor) applyListenOp(op listenOp) {
+	if e.hub == nil || e.subscriber == nil {
+		return
+	}
+	switch op.kind {
+	case listenOpListen:
+		if !e.listens[op.channel] {
+			e.hub.Subscribe(e.subscriber, op.channel)
+			e.listens[op.channel] = true
+		}
+	case listenOpUnlisten:
+		if e.listens[op.channel] {
+			e.hub.Unsubscribe(e.subscriber, op.channel)
+			delete(e.listens, op.channel)
+		}
+	case listenOpUnlistenAll:
+		for channel := range e.listens {
+			e.hub.Unsubscribe(e.subscriber, channel)
+		}
+		e.listens = make(map[string]bool)
+	}
+}
+
+func (e *Executor) publish(n storage.Notification) {
+	if e.hub == nil {
+		return
+	}
+	e.hub.Publish(n)
+}
+
+// finishCommit applies staged subscription changes and releases queued
+// notifications. It runs only after the storage commit has succeeded.
+func (e *Executor) finishCommit(lsn uint64) {
+	for _, op := range e.listenOps {
+		e.applyListenOp(op)
+	}
+	for _, n := range e.notifyQueue {
+		n.LSN = lsn
+		e.publish(n)
+	}
+	e.clearNotificationState()
+}
+
+// finishRollback discards staged subscription changes and notifications.
+func (e *Executor) finishRollback() {
+	e.clearNotificationState()
+}
+
+func (e *Executor) clearNotificationState() {
+	e.listenOps = nil
+	e.notifyQueue = nil
+	e.notificationSaves = nil
+}
+
+func (e *Executor) notificationSnapshot() notificationSnapshot {
+	return notificationSnapshot{
+		listenOps:   len(e.listenOps),
+		notifyQueue: len(e.notifyQueue),
+	}
+}
+
+func (e *Executor) rollbackNotificationsTo(s notificationSnapshot) {
+	if s.listenOps <= len(e.listenOps) {
+		e.listenOps = e.listenOps[:s.listenOps]
+	}
+	if s.notifyQueue <= len(e.notifyQueue) {
+		e.notifyQueue = e.notifyQueue[:s.notifyQueue]
+	}
+}

pkg/executor/notify_test.go

@@ -0,0 +1,196 @@
+package executor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+func newNotifyExec(t *testing.T, hub *storage.Hub, schema *storage.SchemaManager, table *storage.TableManager, pid int32) *Executor {
+	t.Helper()
+	e := newExec(schema, table)
+	e.SetBackendPID(pid)
+	e.SetNotificationHub(hub)
+	t.Cleanup(func() { _ = e.Close() })
+	return e
+}
+
+func recvNotification(t *testing.T, e *Executor) storage.Notification {
+	t.Helper()
+	select {
+	case n := <-e.NotificationChannel():
+		return n
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for notification")
+		return storage.Notification{}
+	}
+}
+
+func expectNoNotification(t *testing.T, e *Executor) {
+	t.Helper()
+	select {
+	case n := <-e.NotificationChannel():
+		t.Fatalf("unexpected notification: %+v", n)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+func TestNotifyDeliversToListener(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN updates")
+	execMust(t, sender, "NOTIFY updates, 'hello'")
+
+	n := recvNotification(t, listener)
+	if n.Channel != "updates" || n.Payload != "hello" || n.PID != 202 {
+		t.Fatalf("unexpected notification: %+v", n)
+	}
+}
+
+func TestListenTakesEffectAtCommit(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "BEGIN")
+	execMust(t, listener, "LISTEN late")
+	execMust(t, sender, "NOTIFY late, 'before'")
+	expectNoNotification(t, listener)
+
+	execMust(t, listener, "COMMIT")
+	execMust(t, sender, "NOTIFY late, 'after'")
+	n := recvNotification(t, listener)
+	if n.Payload != "after" {
+		t.Fatalf("payload = %q, want after", n.Payload)
+	}
+}
+
+func TestListenRollbackDiscardsSubscription(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "BEGIN")
+	execMust(t, listener, "LISTEN rolledback")
+	execMust(t, listener, "ROLLBACK")
+	execMust(t, sender, "NOTIFY rolledback, 'nope'")
+	expectNoNotification(t, listener)
+}
+
+func TestNotifyOnlyDeliveredAfterCommit(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN events")
+	execMust(t, sender, "BEGIN")
+	execMust(t, sender, "NOTIFY events, 'pending'")
+	expectNoNotification(t, listener)
+	execMust(t, sender, "COMMIT")
+
+	n := recvNotification(t, listener)
+	if n.Payload != "pending" || n.LSN == 0 {
+		t.Fatalf("unexpected notification: %+v", n)
+	}
+}
+
+func TestNotifyRollbackDiscards(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN events")
+	execMust(t, sender, "BEGIN")
+	execMust(t, sender, "NOTIFY events, 'gone'")
+	execMust(t, sender, "ROLLBACK")
+	expectNoNotification(t, listener)
+}
+
+func TestNotifySavepointRollback(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN events")
+	execMust(t, sender, "BEGIN")
+	execMust(t, sender, "NOTIFY events, 'kept'")
+	execMust(t, sender, "SAVEPOINT sp")
+	execMust(t, sender, "NOTIFY events, 'discarded'")
+	execMust(t, sender, "ROLLBACK TO SAVEPOINT sp")
+	execMust(t, sender, "COMMIT")
+
+	n := recvNotification(t, listener)
+	if n.Payload != "kept" {
+		t.Fatalf("payload = %q, want kept", n.Payload)
+	}
+	expectNoNotification(t, listener)
+}
+
+func TestNotifyDuplicatesCollapseInTransaction(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN events")
+	execMust(t, sender, "BEGIN")
+	execMust(t, sender, "NOTIFY events, 'same'")
+	execMust(t, sender, "NOTIFY events, 'same'")
+	execMust(t, sender, "NOTIFY events, 'different'")
+	execMust(t, sender, "COMMIT")
+
+	first := recvNotification(t, listener)
+	second := recvNotification(t, listener)
+	if first.Payload != "same" || second.Payload != "different" {
+		t.Fatalf("unexpected notifications: %+v %+v", first, second)
+	}
+	expectNoNotification(t, listener)
+}
+
+func TestUnlistenAllAndDisconnectCleanup(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	hub := storage.NewHub()
+	listener := newNotifyExec(t, hub, schema, table, 101)
+	sender := newNotifyExec(t, hub, schema, table, 202)
+
+	execMust(t, listener, "LISTEN a")
+	execMust(t, listener, "LISTEN b")
+	execMust(t, listener, "UNLISTEN *")
+	execMust(t, sender, "NOTIFY a, 'x'")
+	execMust(t, sender, "NOTIFY b, 'y'")
+	expectNoNotification(t, listener)
+
+	execMust(t, listener, "LISTEN c")
+	_ = listener.Close()
+	execMust(t, sender, "NOTIFY c, 'z'")
+	expectNoNotification(t, listener)
+}
+
+func TestCommitLSNIsMonotonic(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE lsn_test (id INTEGER PRIMARY KEY, v TEXT)")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO lsn_test VALUES (1, 'a')")
+	first := execMust(t, e, "COMMIT")
+	if first.LSN == 0 {
+		t.Fatal("first commit LSN is zero")
+	}
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO lsn_test VALUES (2, 'b')")
+	second := execMust(t, e, "COMMIT")
+	if second.LSN <= first.LSN {
+		t.Fatalf("second commit LSN %d did not advance past %d", second.LSN, first.LSN)
+	}
+}

pkg/executor/result.go

@@ -14,6 +14,7 @@ type Result struct {
 	RowsAffected int64           // Number of rows affected (for INSERT/UPDATE/DELETE)
 	LastInsertID int64           // Last insert ID (for INSERT with AUTOINCREMENT)
 	CommandTag   string          // Command type (SELECT, INSERT, UPDATE, DELETE, etc.)
+	LSN          uint64          // Commit LSN for a committed transaction (0 when none)
 }
 
 // NewResult creates a new empty result.

pkg/httpserver/handler.go

@@ -153,6 +153,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 			Rows:               sanitizeRows(result.Rows),
 			RowsAffected:       result.RowsAffected,
 			LastInsertID:       result.LastInsertID,
+			LSN:                result.LSN,
 			ExecutionTimeMicro: duration.Microseconds(),
 			RowsReturned:       len(result.Rows),
 		}
@@ -266,6 +267,7 @@ func (s *Server) handleExecute(w http.ResponseWriter, r *http.Request) {
 		results = append(results, ExecuteResult{
 			RowsAffected: result.RowsAffected,
 			LastInsertID: result.LastInsertID,
+			LSN:          result.LSN,
 		})
 	}
 

pkg/httpserver/response.go

@@ -35,6 +35,7 @@ type QueryResponse struct {
 	Rows         [][]interface{} `json:"rows"`
 	RowsAffected int64           `json:"rowsAffected"`
 	LastInsertID int64           `json:"lastInsertId"`
+	LSN          uint64          `json:"lsn,omitempty"` // commit LSN for COMMIT
 	QueryPlan    []string        `json:"queryPlan,omitempty"`
 	// Metrics for usage tracking and billing
 	BytesRead          int64 `json:"bytesRead"`          // Total bytes in the result set
@@ -44,8 +45,9 @@ type QueryResponse struct {
 
 // ExecuteResult represents a single execution result.
 type ExecuteResult struct {
-	RowsAffected int64 `json:"rowsAffected"`
-	LastInsertID int64 `json:"lastInsertId"`
+	RowsAffected int64  `json:"rowsAffected"`
+	LastInsertID int64  `json:"lastInsertId"`
+	LSN          uint64 `json:"lsn,omitempty"`
 }
 
 // ExecuteResponse represents a batch execution response.

pkg/httpserver/server.go

@@ -114,6 +114,7 @@ func NewWithDatabaseManager(config *Config, dbManager *storage.DatabaseManager)
 	var defaultSchema *storage.SchemaManager
 	if defaultDB != nil {
 		defaultExec = executor.New(defaultDB.Schema, defaultDB.Table)
+		defaultExec.SetNotificationHub(defaultDB.Hub)
 		defaultExec.SyncCatalog()
 		defaultSchema = defaultDB.Schema
 	}
@@ -220,6 +221,7 @@ func (s *Server) getExecutorForDatabase(dbName string) (*executor.Executor, *sto
 	}
 
 	exec := executor.New(dbInstance.Schema, dbInstance.Table)
+	exec.SetNotificationHub(dbInstance.Hub)
 	exec.SyncCatalog()
 	return exec, dbInstance.Schema, nil
 }

pkg/parser/ast.go

@@ -442,6 +442,34 @@ type ReleaseStmt struct {
 func (s *ReleaseStmt) node()     {}
 func (s *ReleaseStmt) stmtNode() {}
 
+// Notification statements
+
+// ListenStmt represents LISTEN channel.
+type ListenStmt struct {
+	Channel string
+}
+
+func (s *ListenStmt) node()     {}
+func (s *ListenStmt) stmtNode() {}
+
+// UnlistenStmt represents UNLISTEN channel or UNLISTEN *.
+type UnlistenStmt struct {
+	Channel string
+	All     bool
+}
+
+func (s *UnlistenStmt) node()     {}
+func (s *UnlistenStmt) stmtNode() {}
+
+// NotifyStmt represents NOTIFY channel [, payload].
+type NotifyStmt struct {
+	Channel string
+	Payload string
+}
+
+func (s *NotifyStmt) node()     {}
+func (s *NotifyStmt) stmtNode() {}
+
 // Expression types
 
 // BinaryExpr represents a binary expression.

pkg/parser/notify_test.go

@@ -0,0 +1,124 @@
+package parser
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+)
+
+func parseNotifySQL(t *testing.T, sql string) (Statement, error) {
+	t.Helper()
+	return New(lexer.New(sql)).Parse()
+}
+
+func TestParseListen(t *testing.T) {
+	stmt, err := parseNotifySQL(t, `LISTEN orders`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	listen, ok := stmt.(*ListenStmt)
+	if !ok || listen.Channel != "orders" {
+		t.Fatalf("unexpected statement: %#v", stmt)
+	}
+
+	quoted, err := parseNotifySQL(t, `LISTEN "Mixed Case"`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if quoted.(*ListenStmt).Channel != "Mixed Case" {
+		t.Fatalf("quoted channel = %q", quoted.(*ListenStmt).Channel)
+	}
+}
+
+func TestParseUnlisten(t *testing.T) {
+	stmt, err := parseNotifySQL(t, `UNLISTEN orders`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	unlisten := stmt.(*UnlistenStmt)
+	if unlisten.All || unlisten.Channel != "orders" {
+		t.Fatalf("unexpected unlisten: %#v", unlisten)
+	}
+
+	all, err := parseNotifySQL(t, `UNLISTEN *`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !all.(*UnlistenStmt).All {
+		t.Fatal("expected UNLISTEN *")
+	}
+}
+
+func TestParseNotify(t *testing.T) {
+	stmt, err := parseNotifySQL(t, `NOTIFY orders, 'created'`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	notify := stmt.(*NotifyStmt)
+	if notify.Channel != "orders" || notify.Payload != "created" {
+		t.Fatalf("unexpected notify: %#v", notify)
+	}
+
+	empty, err := parseNotifySQL(t, `NOTIFY orders`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if empty.(*NotifyStmt).Payload != "" {
+		t.Fatalf("payload = %q, want empty", empty.(*NotifyStmt).Payload)
+	}
+
+	escaped, err := parseNotifySQL(t, `NOTIFY orders, 'it''s here'`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if escaped.(*NotifyStmt).Payload != "it's here" {
+		t.Fatalf("payload = %q", escaped.(*NotifyStmt).Payload)
+	}
+}
+
+func TestParseNotifyPayloadLimit(t *testing.T) {
+	payload := strings.Repeat("a", 8000)
+	if _, err := parseNotifySQL(t, `NOTIFY ch, '`+payload+`'`); err != nil {
+		t.Fatalf("8000-byte payload rejected: %v", err)
+	}
+	if _, err := parseNotifySQL(t, `NOTIFY ch, '`+payload+`a'`); err == nil {
+		t.Fatal("expected 8001-byte payload to be rejected")
+	}
+}
+
+func TestParseChannelNameLimit(t *testing.T) {
+	channel := strings.Repeat("a", 63)
+	if _, err := parseNotifySQL(t, `LISTEN `+channel); err != nil {
+		t.Fatalf("63-byte channel rejected: %v", err)
+	}
+	if _, err := parseNotifySQL(t, `LISTEN `+channel+`a`); err == nil {
+		t.Fatal("expected 64-byte channel to be rejected")
+	}
+}
+
+func TestParseNotifyInvalid(t *testing.T) {
+	for _, sql := range []string{
+		`LISTEN`,
+		`UNLISTEN`,
+		`NOTIFY ch,`,
+		`NOTIFY ch, 42`,
+		`NOTIFY ch 'payload'`,
+		`LISTEN 42`,
+	} {
+		if _, err := parseNotifySQL(t, sql); err == nil {
+			t.Errorf("expected %q to fail", sql)
+		}
+	}
+}
+
+func TestNotifyIsNotReservedAsColumnName(t *testing.T) {
+	stmt, err := parseNotifySQL(t, `SELECT notify, listen FROM events`)
+	if err != nil {
+		t.Fatalf("contextual keywords should remain usable as columns: %v", err)
+	}
+	sel := stmt.(*SelectStmt)
+	if len(sel.Columns) != 2 {
+		t.Fatalf("columns = %d, want 2", len(sel.Columns))
+	}
+}

pkg/parser/parser.go

@@ -133,6 +133,19 @@ func (p *Parser) parseStatement() (Statement, error) {
 	if p.isWithStart() {
 		return p.parseWithStatement()
 	}
+	// LISTEN, UNLISTEN, and NOTIFY are contextual keywords: they remain valid
+	// identifiers in expressions and only denote a statement at the start of a
+	// statement. That keeps existing columns named "notify" working.
+	if p.curTokenIs(lexer.TokenIdent) {
+		switch strings.ToUpper(p.curToken.Literal) {
+		case "LISTEN":
+			return p.parseListen()
+		case "UNLISTEN":
+			return p.parseUnlisten()
+		case "NOTIFY":
+			return p.parseNotify()
+		}
+	}
 	switch p.curToken.Type {
 	case lexer.TokenSELECT:
 		return p.parseSelect()
@@ -2094,6 +2107,67 @@ func (p *Parser) parseExplain() (*ExplainStmt, error) {
 
 // Transaction statement parsing
 
+// parseChannelName consumes a channel identifier. PostgreSQL channel names are
+// identifiers (quoted or unquoted) and are limited to 63 bytes, matching
+// NAMEDATALEN - 1.
+func (p *Parser) parseChannelName() (string, error) {
+	if !p.curTokenIs(lexer.TokenIdent) {
+		return "", p.curError("expected channel name")
+	}
+	channel := p.curToken.Literal
+	if channel == "" {
+		return "", p.curError("channel name cannot be empty")
+	}
+	if len(channel) > 63 {
+		return "", p.curError("channel name is longer than 63 bytes")
+	}
+	p.nextToken()
+	return channel, nil
+}
+
+func (p *Parser) parseListen() (*ListenStmt, error) {
+	p.nextToken() // consume LISTEN
+	channel, err := p.parseChannelName()
+	if err != nil {
+		return nil, err
+	}
+	return &ListenStmt{Channel: channel}, nil
+}
+
+func (p *Parser) parseUnlisten() (*UnlistenStmt, error) {
+	p.nextToken() // consume UNLISTEN
+	if p.curTokenIs(lexer.TokenStar) {
+		p.nextToken()
+		return &UnlistenStmt{All: true}, nil
+	}
+	channel, err := p.parseChannelName()
+	if err != nil {
+		return nil, err
+	}
+	return &UnlistenStmt{Channel: channel}, nil
+}
+
+func (p *Parser) parseNotify() (*NotifyStmt, error) {
+	p.nextToken() // consume NOTIFY
+	channel, err := p.parseChannelName()
+	if err != nil {
+		return nil, err
+	}
+	stmt := &NotifyStmt{Channel: channel}
+	if p.curTokenIs(lexer.TokenComma) {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenString) {
+			return nil, p.curError("expected notification payload string")
+		}
+		if len(p.curToken.Literal) > 8000 {
+			return nil, p.curError("notification payload is longer than 8000 bytes")
+		}
+		stmt.Payload = p.curToken.Literal
+		p.nextToken()
+	}
+	return stmt, nil
+}
+
 func (p *Parser) parseBegin() (*BeginStmt, error) {
 	stmt := &BeginStmt{}
 

pkg/pgserver/connection.go

@@ -10,10 +10,12 @@ import (
 	"io"
 	"log"
 	"net"
+	"os"
 	"regexp"
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 
 	"github.com/danfragoso/pizzasql-next/pkg/executor"
 	"github.com/danfragoso/pizzasql-next/pkg/lexer"
@@ -38,6 +40,20 @@ type Connection struct {
 	statements     map[string]*preparedStatement
 	portals        map[string]*portal
 	extendedFailed bool
+	backendPID     int32
+	// extendedCycle is true between Parse/Bind/Execute/Describe/Close and the
+	// next Sync. Notifications are not interleaved into an extended-protocol
+	// sequence; they are held until Sync.
+	extendedCycle bool
+}
+
+// backendPIDSeq gives each connection a distinct synthetic backend PID. It is
+// seeded with the process ID so NotificationResponse frames look plausible and
+// do not collide across server processes.
+var backendPIDSeq atomic.Int32
+
+func nextBackendPID() int32 {
+	return int32(os.Getpid()) + backendPIDSeq.Add(1)
 }
 
 type preparedStatement struct {
@@ -62,6 +78,7 @@ func NewConnection(conn net.Conn, dbManager *storage.DatabaseManager, quiet bool
 		portals:    make(map[string]*portal),
 		txStatus:   TxStatusIdle,
 		quiet:      quiet,
+		backendPID: nextBackendPID(),
 	}
 }
 
@@ -69,8 +86,8 @@ func NewConnection(conn net.Conn, dbManager *storage.DatabaseManager, quiet bool
 func (c *Connection) Handle() error {
 	defer func() {
 		if c.executor != nil {
-			if err := c.executor.RollbackActive(); err != nil && !c.quiet {
-				log.Printf("failed to roll back disconnected transaction: %v", err)
+			if err := c.executor.Close(); err != nil && !c.quiet {
+				log.Printf("failed to close session: %v", err)
 			}
 		}
 		c.conn.Close()
@@ -168,7 +185,7 @@ func (c *Connection) Handle() error {
 	}
 
 	// Send backend key data (for cancellation - we don't implement this yet)
-	if err := c.sendBackendKeyData(12345, 67890); err != nil {
+	if err := c.sendBackendKeyData(c.backendPID, 67890); err != nil {
 		return err
 	}
 
@@ -177,25 +194,73 @@ func (c *Connection) Handle() error {
 		return err
 	}
 
-	// Message loop
-	for {
-		msg, err := ReadMessage(c.reader)
-		if err != nil {
-			if err == io.EOF {
-				log.Printf("Connection closed by client")
-				return nil
+	// Read messages on a separate goroutine so an idle listener can receive
+	// asynchronous NotificationResponse frames while blocked waiting for input.
+	// Only this goroutine writes to c.writer, which keeps frame order intact.
+	type readResult struct {
+		msg *Message
+		err error
+	}
+	inbox := make(chan readResult, 16)
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		for {
+			msg, err := ReadMessage(c.reader)
+			select {
+			case inbox <- readResult{msg: msg, err: err}:
+			case <-done:
+				return
+			}
+			if err != nil {
+				return
 			}
-			return fmt.Errorf("failed to read message: %w", err)
+		}
+	}()
+
+	notifications := c.executor.NotificationChannel()
+	overflow := c.executor.NotificationOverflow()
+
+	for {
+		// Do not interleave notifications into an extended-protocol cycle. The
+		// channel is disabled until Sync closes the cycle.
+		var notifyCh <-chan storage.Notification
+		if !c.extendedCycle {
+			notifyCh = notifications
 		}
 
-		if err := c.handleMessage(msg); err != nil {
-			if err == io.EOF {
-				// Normal termination
-				log.Printf("Connection closed normally")
-				return nil
+		select {
+		case res := <-inbox:
+			if res.err != nil {
+				if res.err == io.EOF {
+					log.Printf("Connection closed by client")
+					return nil
+				}
+				return fmt.Errorf("failed to read message: %w", res.err)
 			}
-			log.Printf("Error handling message: %v", err)
-			return err
+			switch res.msg.Type {
+			case MsgParse, MsgBind, MsgDescribe, MsgExecute, MsgClose:
+				c.extendedCycle = true
+			}
+			if err := c.handleMessage(res.msg); err != nil {
+				if err == io.EOF {
+					log.Printf("Connection closed normally")
+					return nil
+				}
+				log.Printf("Error handling message: %v", err)
+				return err
+			}
+
+		case n := <-notifyCh:
+			if err := c.sendNotification(n); err != nil {
+				return err
+			}
+			if err := c.writer.Flush(); err != nil {
+				return err
+			}
+
+		case <-overflow:
+			return fmt.Errorf("notification queue overflow: slow consumer disconnected")
 		}
 	}
 }
@@ -209,6 +274,8 @@ func (c *Connection) initDatabase(dbName string) error {
 
 	c.schema = db.Schema
 	c.executor = executor.New(db.Schema, db.Table)
+	c.executor.SetBackendPID(c.backendPID)
+	c.executor.SetNotificationHub(db.Hub)
 	c.executor.SyncCatalog()
 
 	return nil
@@ -238,7 +305,11 @@ func (c *Connection) handleMessage(msg *Message) error {
 		return c.writer.Flush()
 	case MsgSync:
 		c.extendedFailed = false
-		return c.sendReadyForQuery()
+		c.extendedCycle = false
+		if err := c.sendReadyForQuery(); err != nil {
+			return err
+		}
+		return c.drainNotifications()
 
 	default:
 		log.Printf("Unknown message type: %c (%d)", msg.Type, msg.Type)
@@ -369,6 +440,11 @@ func (c *Connection) handleQuery(msg *Message) error {
 		if err := c.sendResult(result, stmt); err != nil {
 			return err
 		}
+		if result.LSN > 0 {
+			if err := c.sendLSN(result.LSN); err != nil {
+				return err
+			}
+		}
 	}
 
 	return c.sendReadyForQuery()
@@ -559,7 +635,13 @@ func (c *Connection) handleExecute(msg *Message) error {
 		}
 		return c.failExtended(code, fmt.Errorf("execution error: %w", err))
 	}
-	return c.sendResult(result, stmt)
+	if err := c.sendResult(result, stmt); err != nil {
+		return err
+	}
+	if result.LSN > 0 {
+		return c.sendLSN(result.LSN)
+	}
+	return nil
 }
 
 var catalogFilterPattern = regexp.MustCompile(`(?i)\b(table_name|tablename|table_schema|schemaname|constraint_name|indexname)\s*=\s*'((?:''|[^'])*)'`)
@@ -1055,6 +1137,12 @@ func (c *Connection) getCommandTag(stmt parser.Statement, result *executor.Resul
 	case *parser.ReleaseStmt:
 		c.txStatus = TxStatusInBlock
 		return "RELEASE"
+	case *parser.ListenStmt:
+		return "LISTEN"
+	case *parser.UnlistenStmt:
+		return "UNLISTEN"
+	case *parser.NotifyStmt:
+		return "NOTIFY"
 	default:
 		return "OK"
 	}
@@ -1067,6 +1155,38 @@ func (c *Connection) sendAuthenticationOk() error {
 	return c.writeMessage(MsgAuthenticationOk, mb.Bytes())
 }
 
+// drainNotifications writes every notification currently buffered for this
+// session. It is called at protocol-safe points (after Sync/ReadyForQuery).
+func (c *Connection) drainNotifications() error {
+	ch := c.executor.NotificationChannel()
+	for {
+		select {
+		case n := <-ch:
+			if err := c.sendNotification(n); err != nil {
+				return err
+			}
+		default:
+			return c.writer.Flush()
+		}
+	}
+}
+
+// sendNotification writes a NotificationResponse frame. The frame is only
+// written by the connection's main goroutine, never from a publisher.
+func (c *Connection) sendNotification(n storage.Notification) error {
+	mb := NewMessageBuilder()
+	mb.WriteInt32(n.PID)
+	mb.WriteString(n.Channel)
+	mb.WriteString(n.Payload)
+	return c.writeMessage(MsgNotificationResponse, mb.Bytes())
+}
+
+// sendLSN exposes the last commit LSN as a ParameterStatus. Drivers that do
+// not know the key ignore it; subscribers can use it as an ordering cursor.
+func (c *Connection) sendLSN(lsn uint64) error {
+	return c.sendParameterStatus("pizzasql.lsn", strconv.FormatUint(lsn, 10))
+}
+
 // sendParameterStatus sends a parameter status message
 func (c *Connection) sendParameterStatus(name, value string) error {
 	mb := NewMessageBuilder()

pkg/pgserver/notify_test.go

@@ -0,0 +1,298 @@
+package pgserver
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/danfragoso/pizzasql-next/pkg/executor"
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func newNotifyDatabase(t *testing.T) (*storage.DatabaseManager, *storage.DatabaseInstance) {
+	t.Helper()
+	kv := testkv.New(t)
+	pool := kv.Pool(8)
+	t.Cleanup(func() { pool.Close() })
+	dm := storage.NewDatabaseManager(pool, &storage.DatabaseManagerConfig{
+		DefaultDatabase: "pg_notify",
+		AutoCreate:      true,
+	})
+	db, err := dm.GetDatabase("pg_notify")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return dm, db
+}
+
+func queryMessage(sql string) []byte {
+	data := append([]byte(sql), 0)
+	var msg bytes.Buffer
+	msg.WriteByte(MsgQuery)
+	_ = binary.Write(&msg, binary.BigEndian, uint32(len(data)+4))
+	msg.Write(data)
+	return msg.Bytes()
+}
+
+func writeAndReadUntilReady(t *testing.T, client net.Conn, r *bufio.Reader, payload []byte) []*Message {
+	t.Helper()
+	if _, err := client.Write(payload); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	var msgs []*Message
+	for {
+		_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+		m, err := ReadMessage(r)
+		if err != nil {
+			t.Fatalf("read response: %v", err)
+		}
+		msgs = append(msgs, m)
+		if m.Type == MsgReadyForQuery {
+			return msgs
+		}
+	}
+}
+
+func startNotifyConnection(t *testing.T, dm *storage.DatabaseManager) (net.Conn, *bufio.Reader) {
+	t.Helper()
+	server, client := net.Pipe()
+	c := NewConnection(server, dm, true)
+	done := make(chan error, 1)
+	go func() { done <- c.Handle() }()
+	t.Cleanup(func() {
+		client.Close()
+		<-done
+	})
+
+	if _, err := client.Write(startupMessage(map[string]string{"user": "tester", "database": "pg_notify"})); err != nil {
+		t.Fatalf("startup: %v", err)
+	}
+	r := bufio.NewReader(client)
+	for {
+		_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+		m, err := ReadMessage(r)
+		if err != nil {
+			t.Fatalf("read startup: %v", err)
+		}
+		if m.Type == MsgReadyForQuery {
+			return client, r
+		}
+	}
+}
+
+func publishFromNewSession(t *testing.T, db *storage.DatabaseInstance, channel, payload string) {
+	t.Helper()
+	e := executor.New(db.Schema, db.Table)
+	e.SetBackendPID(4242)
+	e.SetNotificationHub(db.Hub)
+	t.Cleanup(func() { _ = e.Close() })
+	l := lexer.New(fmt.Sprintf("NOTIFY %s, '%s'", channel, payload))
+	stmt, err := parser.New(l).Parse()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := e.Execute(stmt); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestIdleListenerReceivesNotification(t *testing.T) {
+	dm, db := newNotifyDatabase(t)
+	client, r := startNotifyConnection(t, dm)
+
+	writeAndReadUntilReady(t, client, r, queryMessage("LISTEN updates"))
+	publishFromNewSession(t, db, "updates", "hello")
+
+	_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+	m, err := ReadMessage(r)
+	if err != nil {
+		t.Fatalf("read notification: %v", err)
+	}
+	if m.Type != MsgNotificationResponse {
+		t.Fatalf("message type = %c, want NotificationResponse", m.Type)
+	}
+	if len(m.Data) < 6 {
+		t.Fatalf("short notification frame: %v", m.Data)
+	}
+	pid := int32(binary.BigEndian.Uint32(m.Data[:4]))
+	if pid != 4242 {
+		t.Fatalf("pid = %d, want 4242", pid)
+	}
+	pos := 4
+	channel, pos, err := readCString(m.Data, pos)
+	if err != nil || channel != "updates" {
+		t.Fatalf("channel = %q err=%v", channel, err)
+	}
+	payload, _, err := readCString(m.Data, pos)
+	if err != nil || payload != "hello" {
+		t.Fatalf("payload = %q err=%v", payload, err)
+	}
+}
+
+func TestExtendedProtocolDeliversNotificationAfterSync(t *testing.T) {
+	dm, db := newNotifyDatabase(t)
+	client, r := startNotifyConnection(t, dm)
+
+	// Parse + Bind + Execute, then Flush. The connection remains in an
+	// extended-protocol cycle, so a notification published now must not be
+	// written until Sync.
+	var buf bytes.Buffer
+	parseBody := []byte("stmt\x00LISTEN ext\x00\x00\x00")
+	buf.WriteByte(MsgParse)
+	_ = binary.Write(&buf, binary.BigEndian, uint32(len(parseBody)+4))
+	buf.Write(parseBody)
+
+	bindBody := []byte("portal\x00stmt\x00\x00\x00\x00\x00\x00\x00")
+	buf.WriteByte(MsgBind)
+	_ = binary.Write(&buf, binary.BigEndian, uint32(len(bindBody)+4))
+	buf.Write(bindBody)
+
+	execBody := []byte("portal\x00\x00\x00\x00\x00")
+	buf.WriteByte(MsgExecute)
+	_ = binary.Write(&buf, binary.BigEndian, uint32(len(execBody)+4))
+	buf.Write(execBody)
+
+	buf.WriteByte(MsgFlush)
+	_ = binary.Write(&buf, binary.BigEndian, uint32(4))
+	if _, err := client.Write(buf.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+
+	for {
+		_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+		m, err := ReadMessage(r)
+		if err != nil {
+			t.Fatalf("read extended response: %v", err)
+		}
+		if m.Type == MsgCommandComplete {
+			break
+		}
+		if m.Type == MsgErrorResponse {
+			t.Fatalf("extended protocol error: %+v", m.Data)
+		}
+	}
+
+	publishFromNewSession(t, db, "ext", "during-extended")
+
+	// The notification is queued in the subscriber but must not be written
+	// before Sync closes the cycle.
+	_ = client.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
+	if m, err := ReadMessage(r); err == nil {
+		t.Fatalf("notification interleaved into extended cycle: type %c", m.Type)
+	}
+
+	var syncMsg bytes.Buffer
+	syncMsg.WriteByte(MsgSync)
+	_ = binary.Write(&syncMsg, binary.BigEndian, uint32(4))
+	if _, err := client.Write(syncMsg.Bytes()); err != nil {
+		t.Fatal(err)
+	}
+
+	var sawNotification bool
+	for {
+		_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+		m, err := ReadMessage(r)
+		if err != nil {
+			t.Fatalf("read sync response: %v", err)
+		}
+		if m.Type == MsgNotificationResponse {
+			sawNotification = true
+		}
+		if m.Type == MsgReadyForQuery {
+			break
+		}
+	}
+	if !sawNotification {
+		// Notifications are emitted after ReadyForQuery, so the frame follows the
+		// Ready message that closed the extended cycle.
+		_ = client.SetReadDeadline(time.Now().Add(3 * time.Second))
+		m, err := ReadMessage(r)
+		if err != nil {
+			t.Fatalf("notification was not delivered after Sync: %v", err)
+		}
+		if m.Type != MsgNotificationResponse {
+			t.Fatalf("post-Sync message type = %c, want NotificationResponse", m.Type)
+		}
+	}
+}
+
+func TestCommitLSNParameterStatus(t *testing.T) {
+	dm, _ := newNotifyDatabase(t)
+	client, r := startNotifyConnection(t, dm)
+
+	writeAndReadUntilReady(t, client, r, queryMessage("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)"))
+	writeAndReadUntilReady(t, client, r, queryMessage("BEGIN"))
+	writeAndReadUntilReady(t, client, r, queryMessage("INSERT INTO t VALUES (1, 'a')"))
+	msgs := writeAndReadUntilReady(t, client, r, queryMessage("COMMIT"))
+
+	found := false
+	for _, m := range msgs {
+		if m.Type != MsgParameterStatus {
+			continue
+		}
+		name, pos, err := readCString(m.Data, 0)
+		if err != nil || name != "pizzasql.lsn" {
+			continue
+		}
+		value, _, err := readCString(m.Data, pos)
+		if err != nil || value == "" || value == "0" {
+			t.Fatalf("invalid lsn value %q", value)
+		}
+		found = true
+	}
+	if !found {
+		t.Fatal("COMMIT did not expose pizzasql.lsn")
+	}
+}
+
+func TestNotificationFrameEncoding(t *testing.T) {
+	c, _ := newTestConnection(t)
+	var out bytes.Buffer
+	c.writer = bufio.NewWriter(&out)
+	if err := c.sendNotification(storage.Notification{PID: 7, Channel: "ch", Payload: "p"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.writer.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	m, err := ReadMessage(bytes.NewReader(out.Bytes()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if m.Type != MsgNotificationResponse {
+		t.Fatalf("type = %c", m.Type)
+	}
+	channel, pos, _ := readCString(m.Data, 4)
+	payload, _, _ := readCString(m.Data, pos)
+	if channel != "ch" || payload != "p" {
+		t.Fatalf("decoded %q %q", channel, payload)
+	}
+}
+
+func TestHandleStartupDrainsNotificationsOnDisconnect(t *testing.T) {
+	dm, db := newNotifyDatabase(t)
+	client, r := startNotifyConnection(t, dm)
+	writeAndReadUntilReady(t, client, r, queryMessage("LISTEN cleanup"))
+	if len(db.Hub.Channels()) != 1 {
+		t.Fatalf("channels = %v", db.Hub.Channels())
+	}
+	// Terminate and give Handle time to run its cleanup defer.
+	if _, err := client.Write([]byte{MsgTerminate, 0, 0, 0, 4}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if len(db.Hub.Channels()) == 0 {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("subscription not cleaned up: %v", db.Hub.Channels())
+}

pkg/storage/db_manager.go

@@ -10,6 +10,10 @@ type DatabaseInstance struct {
 	Name   string
 	Schema *SchemaManager
 	Table  *TableManager
+	// Hub is the shared LISTEN/NOTIFY channel registry for this database. All
+	// connections to the same database instance publish and subscribe through
+	// the same hub; separate instances (tenants) are fully isolated.
+	Hub *Hub
 }
 
 // DatabaseManager manages multiple database instances.
@@ -89,6 +93,7 @@ func (dm *DatabaseManager) getOrCreateDatabase(name string) *DatabaseInstance {
 		Name:   name,
 		Schema: schema,
 		Table:  table,
+		Hub:    NewHub(),
 	}
 
 	dm.databases[name] = db

pkg/storage/notify.go

@@ -0,0 +1,218 @@
+package storage
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+// Notification is a single NOTIFY event delivered to subscribers of a channel.
+// Payload is the application-supplied text (at most 8000 bytes), PID is the
+// backend that raised it, and LSN is the commit LSN of the transaction that
+// made it visible (zero for a NOTIFY outside a write transaction).
+type Notification struct {
+	PID     int32
+	Channel string
+	Payload string
+	LSN     uint64
+}
+
+// Subscriber is one connection's view of the notification hub. A subscriber
+// may listen to many channels and is registered with exactly one Hub.
+type Subscriber struct {
+	id int32
+
+	// notifications is bounded so a slow client can never block a publisher.
+	// Hub.Publish drops the whole subscriber (closing overflow) instead of
+	// waiting when the buffer is full.
+	notifications chan Notification
+	overflow      chan struct{}
+	overflowOnce  sync.Once
+}
+
+// NewSubscriber creates a subscriber with the given backend PID and an
+// outbound queue of queueSize notifications.
+func NewSubscriber(pid int32, queueSize int) *Subscriber {
+	if queueSize <= 0 {
+		queueSize = defaultNotificationQueue
+	}
+	return &Subscriber{
+		id:            pid,
+		notifications: make(chan Notification, queueSize),
+		overflow:      make(chan struct{}),
+	}
+}
+
+const defaultNotificationQueue = 128
+
+// Notifications returns the channel notifications are delivered on.
+func (s *Subscriber) Notifications() <-chan Notification { return s.notifications }
+
+// Overflow is closed when the subscriber could not keep up and has been
+// removed from the hub. Protocol layers should disconnect the connection.
+func (s *Subscriber) Overflow() <-chan struct{} { return s.overflow }
+
+// PID returns the backend PID associated with the subscriber.
+func (s *Subscriber) PID() int32 { return s.id }
+
+func (s *Subscriber) markOverflow() {
+	s.overflowOnce.Do(func() { close(s.overflow) })
+}
+
+// Hub is a per-database channel registry. It maps channel names to subscribed
+// connections and broadcasts notifications without blocking publishers. A Hub
+// is scoped to one DatabaseInstance, so tenants sharing a process never observe
+// each other's channels.
+type Hub struct {
+	mu          sync.RWMutex
+	channels    map[string]map[*Subscriber]struct{}
+	subscribers map[*Subscriber]struct{}
+	published   atomic.Uint64
+	dropped     atomic.Uint64
+	lastLSN     atomic.Uint64
+}
+
+// NewHub creates an empty notification hub.
+func NewHub() *Hub {
+	return &Hub{
+		channels:    make(map[string]map[*Subscriber]struct{}),
+		subscribers: make(map[*Subscriber]struct{}),
+	}
+}
+
+// Subscribe registers a subscriber for a channel. Duplicate subscriptions are
+// idempotent.
+func (h *Hub) Subscribe(s *Subscriber, channel string) {
+	if h == nil || s == nil || channel == "" {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	subs := h.channels[channel]
+	if subs == nil {
+		subs = make(map[*Subscriber]struct{})
+		h.channels[channel] = subs
+	}
+	subs[s] = struct{}{}
+	h.subscribers[s] = struct{}{}
+}
+
+// Unsubscribe removes a subscriber from one channel.
+func (h *Hub) Unsubscribe(s *Subscriber, channel string) {
+	if h == nil || s == nil {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.unsubscribeLocked(s, channel)
+}
+
+// UnsubscribeAll removes a subscriber from every channel. It is safe to call
+// more than once (disconnect cleanup may race an overflow removal).
+func (h *Hub) UnsubscribeAll(s *Subscriber) {
+	if h == nil || s == nil {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for channel := range h.channels {
+		h.unsubscribeLocked(s, channel)
+	}
+	delete(h.subscribers, s)
+}
+
+func (h *Hub) unsubscribeLocked(s *Subscriber, channel string) {
+	subs := h.channels[channel]
+	if subs == nil {
+		return
+	}
+	delete(subs, s)
+	if len(subs) == 0 {
+		delete(h.channels, channel)
+	}
+	// The subscriber remains in h.subscribers if it still listens elsewhere.
+	remaining := false
+	for _, members := range h.channels {
+		if _, ok := members[s]; ok {
+			remaining = true
+			break
+		}
+	}
+	if !remaining {
+		delete(h.subscribers, s)
+	}
+}
+
+// Publish delivers a notification to every subscriber of its channel. It never
+// blocks: a subscriber whose bounded queue is full is removed and marked
+// overflowed so its protocol layer can disconnect it.
+func (h *Hub) Publish(n Notification) {
+	if h == nil || n.Channel == "" {
+		return
+	}
+	n.LSN = h.assignLSN(n.LSN)
+	h.mu.RLock()
+	targets := make([]*Subscriber, 0, len(h.channels[n.Channel]))
+	for s := range h.channels[n.Channel] {
+		targets = append(targets, s)
+	}
+	h.mu.RUnlock()
+
+	if len(targets) == 0 {
+		return
+	}
+
+	var overflowed []*Subscriber
+	for _, s := range targets {
+		select {
+		case s.notifications <- n:
+			h.published.Add(1)
+		default:
+			s.markOverflow()
+			overflowed = append(overflowed, s)
+			h.dropped.Add(1)
+		}
+	}
+
+	for _, s := range overflowed {
+		h.UnsubscribeAll(s)
+	}
+}
+
+// assignLSN makes notification ordering monotonic within a database. A write
+// transaction supplies its storage LSN; a NOTIFY-only transaction (which has no
+// durable write) advances the hub's logical cursor instead of reporting zero.
+func (h *Hub) assignLSN(lsn uint64) uint64 {
+	for {
+		current := h.lastLSN.Load()
+		next := current + 1
+		if lsn > next {
+			next = lsn
+		}
+		if h.lastLSN.CompareAndSwap(current, next) {
+			return next
+		}
+	}
+}
+
+// Stats returns the number of published and dropped notifications. It is used
+// by metrics and tests.
+func (h *Hub) Stats() (published, dropped uint64) {
+	if h == nil {
+		return 0, 0
+	}
+	return h.published.Load(), h.dropped.Load()
+}
+
+// Channels returns the channels with at least one subscriber.
+func (h *Hub) Channels() []string {
+	if h == nil {
+		return nil
+	}
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	channels := make([]string, 0, len(h.channels))
+	for channel := range h.channels {
+		channels = append(channels, channel)
+	}
+	return channels
+}

pkg/storage/notify_test.go

@@ -0,0 +1,90 @@
+package storage
+
+import (
+	"testing"
+	"time"
+)
+
+func TestHubSubscribePublishUnsubscribe(t *testing.T) {
+	h := NewHub()
+	a := NewSubscriber(1, 4)
+	b := NewSubscriber(2, 4)
+	h.Subscribe(a, "orders")
+	h.Subscribe(a, "orders") // idempotent
+	h.Subscribe(b, "other")
+
+	h.Publish(Notification{PID: 9, Channel: "orders", Payload: "one"})
+	select {
+	case n := <-a.Notifications():
+		if n.Payload != "one" || n.LSN == 0 {
+			t.Fatalf("unexpected notification: %+v", n)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("notification not delivered")
+	}
+	select {
+	case n := <-b.Notifications():
+		t.Fatalf("received notification for another channel: %+v", n)
+	default:
+	}
+
+	h.Unsubscribe(a, "orders")
+	h.Publish(Notification{Channel: "orders", Payload: "two"})
+	select {
+	case n := <-a.Notifications():
+		t.Fatalf("received notification after unsubscribe: %+v", n)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+func TestHubSlowConsumerIsDropped(t *testing.T) {
+	h := NewHub()
+	s := NewSubscriber(1, 2)
+	h.Subscribe(s, "ch")
+
+	for i := 0; i < 5; i++ {
+		h.Publish(Notification{Channel: "ch", Payload: "x"})
+	}
+	select {
+	case <-s.Overflow():
+	case <-time.After(time.Second):
+		t.Fatal("overflow signal not closed")
+	}
+	if channels := h.Channels(); len(channels) != 0 {
+		t.Fatalf("slow consumer still subscribed: %v", channels)
+	}
+	published, dropped := h.Stats()
+	if published != 2 || dropped == 0 {
+		t.Fatalf("published=%d dropped=%d", published, dropped)
+	}
+}
+
+func TestHubsAreIsolated(t *testing.T) {
+	one := NewHub()
+	two := NewHub()
+	s := NewSubscriber(1, 2)
+	one.Subscribe(s, "ch")
+	two.Publish(Notification{Channel: "ch", Payload: "nope"})
+	select {
+	case n := <-s.Notifications():
+		t.Fatalf("cross-hub notification: %+v", n)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+func TestNotificationLSNMonotonicAcrossNotifyOnlyPublishes(t *testing.T) {
+	h := NewHub()
+	s := NewSubscriber(1, 8)
+	h.Subscribe(s, "ch")
+	h.Publish(Notification{Channel: "ch", Payload: "a"})
+	h.Publish(Notification{Channel: "ch", Payload: "b", LSN: 1000})
+	h.Publish(Notification{Channel: "ch", Payload: "c"})
+
+	var got []uint64
+	for i := 0; i < 3; i++ {
+		got = append(got, (<-s.Notifications()).LSN)
+	}
+	if got[0] == 0 || got[1] != 1000 || got[2] <= got[1] {
+		t.Fatalf("lsns = %v", got)
+	}
+}

pkg/storage/tx.go

@@ -148,16 +148,25 @@ func (s *Session) Rollback() error {
 }
 
 // Commit validates and durably applies the staged transaction in one atomic
-// compare-and-swap batch write.
+// compare-and-swap batch write. It discards the commit LSN; callers that need a
+// ordering cursor use CommitWithLSN.
 func (s *Session) Commit() error {
+	_, err := s.CommitWithLSN()
+	return err
+}
+
+// CommitWithLSN behaves like Commit and additionally returns the LSN assigned
+// to the durable batch. A read-only or empty transaction returns zero because
+// no new durable state was published.
+func (s *Session) CommitWithLSN() (uint64, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if !s.inTx {
-		return fmt.Errorf("cannot commit: no transaction in progress")
+		return 0, fmt.Errorf("cannot commit: no transaction in progress")
 	}
 	if s.aborted {
 		s.resetLocked()
-		return fmt.Errorf("current transaction is aborted")
+		return 0, fmt.Errorf("current transaction is aborted")
 	}
 
 	// Acquire the per-table gates, escalating any table whose UNIQUE index
@@ -165,7 +174,7 @@ func (s *Session) Commit() error {
 	_, _, unlockAll, uniqueTables, err := s.acquireCommitLocks()
 	if err != nil {
 		s.resetLocked()
-		return err
+		return 0, err
 	}
 	defer unlockAll()
 
@@ -173,13 +182,13 @@ func (s *Session) Commit() error {
 	for t, gen := range s.scanGens {
 		if s.table.generation(t) != gen {
 			s.resetLocked()
-			return ErrSerialization
+			return 0, ErrSerialization
 		}
 	}
 	for key, predicate := range s.predicateGens {
 		if s.table.predicateGeneration(key) != predicate.gen {
 			s.resetLocked()
-			return ErrSerialization
+			return 0, ErrSerialization
 		}
 	}
 
@@ -203,7 +212,7 @@ func (s *Session) Commit() error {
 		}
 		if err := s.table.validateUniqueRows(t, pending, excluded); err != nil {
 			s.resetLocked()
-			return err
+			return 0, err
 		}
 	}
 
@@ -224,7 +233,7 @@ func (s *Session) Commit() error {
 				data, err := encodeRow(e.row)
 				if err != nil {
 					s.resetLocked()
-					return err
+					return 0, err
 				}
 				ops = append(ops, BatchOp{Op: batchPut, Key: []byte(key), Value: data})
 			}
@@ -235,22 +244,24 @@ func (s *Session) Commit() error {
 	// A read-only transaction has nothing to write; commit trivially.
 	if len(ops) == 0 {
 		s.resetLocked()
-		return nil
+		return 0, nil
 	}
 
 	var committed bool
+	var commitLSN uint64
 	err = s.table.pool.WithClient(func(c *KVClient) error {
-		_, ok, err := c.CompareBatchWrite(checks, ops, nil)
+		lsn, ok, err := c.CompareBatchWrite(checks, ops, nil)
+		commitLSN = lsn
 		committed = ok
 		return err
 	})
 	if err != nil {
 		s.resetLocked()
-		return err
+		return 0, err
 	}
 	if !committed {
 		s.resetLocked()
-		return ErrSerialization
+		return 0, ErrSerialization
 	}
 
 	// Advance generations and invalidate derived caches for written tables.
@@ -261,7 +272,7 @@ func (s *Session) Commit() error {
 	}
 
 	s.resetLocked()
-	return nil
+	return commitLSN, nil
 }
 
 // acquireCommitLocks acquires the per-table gates needed to commit the staged