refactor: use pglogrepl's message parser for WAL parsing

Replace custom BinaryParser with pglogrepl.Parse() for consistency
with the upstream library and improved maintainability.

Changes:
- Replace BinaryParser with Parser using pglogrepl.Parse()
- Update WAL struct to use pglogrepl.LSN and uint32 for relationID
- Update CreateActionData to accept *pglogrepl.TupleData
- Remove custom message types from protocol.go (now uses pglogrepl types)
- Rewrite parser tests for new implementation
- Update wal tests for new types
- Fix main.go to use NewParser() instead of NewBinaryParser()
- Fix NatsCredPath validation (not required for all publisher types)
- Fix config test expectations for lowercase field names
- Fix nats_test.go subject name expectation
- Update copilot-instructions.md to document pglogrepl usage

Signed-off-by: Tamal Saha <[email protected]>

Author: tamalsaha

.github/copilot-instructions.md

@@ -2,18 +2,20 @@
 
 ## Architecture & Data Flow
 
-**pgoutbox** captures PostgreSQL changes via logical decoding and publishes to message brokers (NATS, Kafka, RabbitMQ, Google Pub/Sub).
+**pgoutbox** captures PostgreSQL changes via logical decoding and publishes to message brokers (NATS JetStream, Kafka, RabbitMQ, Google Pub/Sub).
 
 ```
 PostgreSQL WAL → pgx replication → Parser.ParseWalMessage() → WAL object → filter → Event → Publisher.Publish() → broker
 ```
 
 ### Key Components
-- [internal/listener/listener.go](../internal/listener/listener.go) - Replication orchestration, slot/publication lifecycle
-- [internal/listener/transaction/parser.go](../internal/listener/transaction/parser.go) - WAL binary protocol decoding
+- [internal/listener/listener.go](../internal/listener/listener.go) - Replication orchestration, slot/publication lifecycle, health probes
+- [internal/listener/transaction/parser.go](../internal/listener/transaction/parser.go) - WAL message parsing using `pglogrepl.Parse()` (`pgoutput` plugin)
+- [internal/listener/transaction/wal.go](../internal/listener/transaction/wal.go) - Transaction state, event filtering via `CreateEventsWithFilter()`
 - [internal/publisher/](../internal/publisher/) - Broker adapters implementing `eventPublisher` interface
 - [apis/config.go](../apis/config.go) - Config types with Viper loading (env prefix: `WAL_`)
 - [apis/event.go](../apis/event.go) - Event struct with `SubjectName()` for topic construction
+- [cmd/pgoutbox/init.go](../cmd/pgoutbox/init.go) - Publisher factory, connection setup with retry
 
 ## Developer Workflows
 
@@ -25,48 +27,63 @@ make ci             # verify + lint + build + unit-tests
 make fmt            # Format source code
 ```
 
-All commands run in Docker (`ghcr.io/appscode/golang-dev:1.24`). Output: `bin/pgoutbox-{OS}-{ARCH}`.
+All commands run in Docker (`ghcr.io/appscode/golang-dev:1.25`). Output: `bin/pgoutbox-{OS}-{ARCH}`.
 
 **Local PostgreSQL**: `docker/docker-compose.yml` + `docker/scripts/` for test DB setup.
 
 ## Critical Patterns
 
-### Interface-Based Mocking
-Listener depends on interfaces (`eventPublisher`, `parser`, `replication`, `repository`). Mock implementations in `*_mock_test.go` files:
+### Interface-Based Testing
+Listener depends on four interfaces defined in [listener.go](../internal/listener/listener.go):
 ```go
-// internal/listener/listener.go
-type eventPublisher interface {
-    Publish(context.Context, string, *apis.Event) error
-}
+type eventPublisher interface { Publish(context.Context, string, *apis.Event) error }
+type parser interface { ParseWalMessage([]byte, *tx.WAL) error }
+type replication interface { CreateReplicationSlot(...); StartReplication(...); ReceiveMessage(...) }
+type repository interface { CreatePublication(...); GetSlotLSN(...); IsReplicationActive(...) }
 ```
+Mock implementations in `*_mock_test.go` files use `github.com/stretchr/testify/mock`.
+
+### Adding a New Publisher
+1. Create `internal/publisher/{broker}.go` implementing `Publish(ctx, subject, *Event) error` and `Close() error`
+2. Add type constant in [apis/config.go](../apis/config.go): `PublisherType{Name} PublisherType = "{name}"`
+3. Add case in `factoryPublisher()` in [cmd/pgoutbox/init.go](../cmd/pgoutbox/init.go)
 
 ### Configuration
-- YAML config + environment overrides (prefix `WAL_`, e.g., `WAL_DATABASE_HOST`)
-- Struct tags: both `mapstructure` and `json` required
+- YAML config + environment overrides (prefix `WAL_`, e.g., `WAL_DATABASE_HOST`, `WAL_LISTENER_SLOTNAME`)
+- Struct tags: both `mapstructure` and `json` required on all config fields
 - Publisher types: `nats`, `kafka`, `rabbitmq`, `google_pubsub`
 
 ### Topic Naming
 Format: `{publisher.topic}.{publisher.topicPrefix}schemas.{schema}.tables.{table}`
-Override via `listener.topicsMap` in config.
+Override specific topics via `listener.topicsMap` in config.
+
+### WAL Message Parsing
+Uses `github.com/jackc/pglogrepl` for WAL message parsing. The `Parser.ParseWalMessage()` method uses `pglogrepl.Parse()` to decode messages:
+- Message types: `BeginMessage`, `CommitMessage`, `RelationMessage`, `InsertMessage`, `UpdateMessage`, `DeleteMessage`
+- Types: `pglogrepl.LSN` for LSN values, `uint32` for relation IDs, `*pglogrepl.TupleData` for row data
+- Column flags: `Flags == 1` indicates the column is part of the primary key
 
 ### Logging
 Use `log/slog` with injected `*slog.Logger`:
 ```go
 l.log.Debug("message", slog.String("key", value))
+l.log.Info("event was sent", slog.String("subject", name), slog.String("table", table))
 ```
 
 ## Common Tasks
 
 | Task | Files to Modify |
 |------|-----------------|
-| Add publisher | `internal/publisher/{broker}.go` (implement `eventPublisher`), `cmd/pgoutbox/init.go` |
-| Add filter logic | `apis/config.go` (`FilterStruct`), `internal/listener/transaction/parser.go` |
-| Extend WAL parsing | `internal/listener/transaction/parser.go`, `transaction/data.go` |
-| Add metrics | `apis/metrics.go` (Prometheus counters) |
+| Add publisher | `internal/publisher/{broker}.go`, `apis/config.go` (type const), `cmd/pgoutbox/init.go` (factory) |
+| Add filter logic | `apis/config.go` (`FilterStruct`), `internal/listener/transaction/wal.go` (`CreateEventsWithFilter`) |
+| Extend WAL parsing | `internal/listener/transaction/parser.go`, `transaction/data.go` (uses `pglogrepl.Parse()`) |
+| Add metrics | `apis/metrics.go` (Prometheus counters), update `monitor` interface |
+| Add config field | `apis/config.go` (add both `mapstructure` and `json` tags) |
 
 ## Standards
 
 - **License**: Apache 2.0 headers required (see `hack/license/`)
 - **Errors**: Wrap with `fmt.Errorf("context: %w", err)`
-- **Publication name**: Hardcoded as `_pgoutbox_`
+- **Publication name**: Hardcoded as `pgoutbox` (auto-created for all tables)
 - **Replication slot**: Configured via `listener.slotName`, auto-created on startup
+- **Tests**: Table-driven tests with `testify/assert` and `testify/mock`

.github/workflows/ci.yml

@@ -1,25 +1,33 @@
-name: ci
+name: CI
 
 on:
-  push:
-    branches: [ "master" ]
   pull_request:
-    branches: [ "master" ]
+    branches:
+      - '*'
+  push:
+    branches:
+      - master
+  workflow_dispatch:
 
-jobs:
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.ref }}
+  cancel-in-progress: true
 
+jobs:
   build:
+    name: Build
     runs-on: ubuntu-24.04
     steps:
-    - uses: actions/checkout@v4
+      - name: Set up Go 1.25
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.25'
+        id: go
 
-    - name: Set up Go
-      uses: actions/setup-go@v5
-      with:
-        go-version: '1.25'
+      - uses: actions/checkout@v4
 
-    - name: Build
-      run: go build -v ./...
+      - name: Build
+        run: go build -v ./...
 
-    - name: Test
-      run: go test -v ./...
+      - name: Test
+        run: go test -v ./...

apis/config.go

@@ -41,7 +41,7 @@ type ListenerCfg struct {
 type PublisherCfg struct {
 	Type            PublisherType `valid:"required" json:"type" mapstructure:"type"`
 	Address         string        `valid:"required" json:"address" mapstructure:"address"`
-	NatsCredPath    string        `valid:"required" json:"natsCredPath" mapstructure:"natsCredPath"`
+	NatsCredPath    string        `json:"natsCredPath" mapstructure:"natsCredPath"`
 	Topic           string        `valid:"required" json:"topic" mapstructure:"topic"`
 	TopicPrefix     string        `json:"topicPrefix" mapstructure:"topicPrefix"`
 	EnableTLS       bool          `json:"enableTLS" mapstructure:"enableTlS"`

apis/config_test.go

@@ -39,11 +39,10 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:         "kafka",
-					Address:      "addr",
-					Topic:        "stream",
-					TopicPrefix:  "prefix",
-					NatsCredPath: "/etc/nats/creds/admin.creds",
+					Type:        "kafka",
+					Address:     "addr",
+					Topic:       "stream",
+					TopicPrefix: "prefix",
 				},
 			},
 			wantErr: nil,
@@ -66,11 +65,10 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:         "kafka",
-					Address:      "addr",
-					Topic:        "stream",
-					TopicPrefix:  "prefix",
-					NatsCredPath: "/etc/nats/creds/admin.creds",
+					Type:        "kafka",
+					Address:     "addr",
+					Topic:       "stream",
+					TopicPrefix: "prefix",
 				},
 			},
 			wantErr: errors.New("Listener.refreshConnection: non zero value required;Listener.slotName: non zero value required"),
@@ -93,11 +91,10 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Type:         "kafka",
-					Address:      "addr",
-					Topic:        "stream",
-					TopicPrefix:  "prefix",
-					NatsCredPath: "/etc/nats/creds/admin.creds",
+					Type:        "kafka",
+					Address:     "addr",
+					Topic:       "stream",
+					TopicPrefix: "prefix",
 				},
 			},
 			wantErr: errors.New("Database.host: non zero value required;Database.port: non zero value required"),
@@ -122,9 +119,8 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Publisher: &PublisherCfg{
-					Topic:        "stream",
-					TopicPrefix:  "prefix",
-					NatsCredPath: "/etc/nats/creds/admin.creds",
+					Topic:       "stream",
+					TopicPrefix: "prefix",
 				},
 			},
 			wantErr: errors.New("Publisher.address: non zero value required;Publisher.type: non zero value required"),

cmd/pgoutbox/main.go

@@ -17,7 +17,6 @@ limitations under the License.
 package main
 
 import (
-	"encoding/binary"
 	"fmt"
 	"log/slog"
 	"os"
@@ -110,7 +109,7 @@ func main() {
 				listener.NewRepository(pgxConn),
 				newReplicationConn(pgConn),
 				pub,
-				transaction.NewBinaryParser(logger, binary.BigEndian),
+				transaction.NewParser(logger),
 				apis.NewMetrics(),
 			)