### Run pglogrepl_demo

Source: https://github.com/jackc/pglogrepl/blob/master/example/pglogrepl_demo/README.md

Start the pglogrepl_demo application by setting the connection string environment variable and running the Go program. Ensure the database user has superuser privileges.

```bash
$ PGLOGREPL_DEMO_CONN_STRING="postgres://pglogrepl:secret@127.0.0.1/pglogrepl?replication=database" go run main.go
```

--------------------------------

### Start Logical Replication Stream

Source: https://context7.com/jackc/pglogrepl/llms.txt

Initiates logical replication by sending the START_REPLICATION command. The connection enters copy-both mode to receive WAL data. Ensure the slot exists and is configured correctly.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	slotName := "my_slot"

	// Get current position to start from
	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
	if err != nil {
		log.Fatalf("IdentifySystem failed: %v", err)
	}

	// Start logical replication with pgoutput plugin options
	// Protocol version 2 enables streaming of large transactions (PG 14+)
	err = pglogrepl.StartReplication(
		context.Background(),
		conn,
		slotName,
		sysident.XLogPos,
		pglogrepl.StartReplicationOptions{
			Mode: pglogrepl.LogicalReplication,
			PluginArgs: []string{
				"proto_version '2'",
				"publication_names 'my_publication'",
				"messages 'true'",
				"streaming 'true'",
			},
		},
	)
	if err != nil {
		log.Fatalf("StartReplication failed: %v", err)
	}

	log.Println("Logical replication started successfully")
	// Connection is now in copy-both mode, ready to receive messages
}
```

--------------------------------

### Complete Logical Replication Example in Go

Source: https://context7.com/jackc/pglogrepl/llms.txt

This Go program sets up and consumes logical replication changes. It requires a PostgreSQL connection string with replication enabled and demonstrates creating a publication, establishing a replication slot, and processing various message types including relation, insert, update, and delete messages.

```go
package main

import (
	"context"
	"log"
	"os"
	"time"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
)

func main() {
	const outputPlugin = "pgoutput"
	const slotName = "my_replication_slot"
	const publicationName = "my_publication"

	// Connect with replication=database
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Create publication (run once, can also be done via regular SQL)
	conn.Exec(context.Background(), "DROP PUBLICATION IF EXISTS "+publicationName)
	conn.Exec(context.Background(), "CREATE PUBLICATION "+publicationName+" FOR ALL TABLES")

	// Get system info and starting position
	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
	if err != nil {
		log.Fatalf("IdentifySystem failed: %v", err)
	}
	log.Printf("Connected to %s at %s", sysident.DBName, sysident.XLogPos)

	// Create temporary replication slot
	_, err = pglogrepl.CreateReplicationSlot(
		context.Background(), conn, slotName, outputPlugin,
		pglogrepl.CreateReplicationSlotOptions{Temporary: true},
	)
	if err != nil {
		log.Fatalf("CreateReplicationSlot failed: %v", err)
	}

	// Start replication with V2 protocol (streaming support)
	err = pglogrepl.StartReplication(
		context.Background(), conn, slotName, sysident.XLogPos,
		pglogrepl.StartReplicationOptions{
			PluginArgs: []string{
				"proto_version '2'",
				"publication_names '" + publicationName + "'",
				"messages 'true'",
				"streaming 'true'",
			},
		},
	)
	if err != nil {
		log.Fatalf("StartReplication failed: %v", err)
	}
	log.Println("Replication started")

	// Main replication loop
	clientXLogPos := sysident.XLogPos
	standbyTimeout := 10 * time.Second
	nextStatusDeadline := time.Now().Add(standbyTimeout)
	relations := make(map[uint32]*pglogrepl.RelationMessageV2)
	inStream := false

	for {
		// Send periodic status updates
		if time.Now().After(nextStatusDeadline) {
			pglogrepl.SendStandbyStatusUpdate(context.Background(), conn,
				pglogrepl.StandbyStatusUpdate{WALWritePosition: clientXLogPos})
			nextStatusDeadline = time.Now().Add(standbyTimeout)
		}

		// Receive with timeout
		ctx, cancel := context.WithDeadline(context.Background(), nextStatusDeadline)
		rawMsg, err := conn.ReceiveMessage(ctx)
		cancel()

		if err != nil {
			if pgconn.Timeout(err) {
				continue
			}
			log.Fatalf("ReceiveMessage failed: %v", err)
		}

		msg, ok := rawMsg.(*pgproto3.CopyData)
		if !ok {
			continue
		}

		switch msg.Data[0] {
		case pglogrepl.PrimaryKeepaliveMessageByteID:
			pkm, _ := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
			if pkm.ServerWALEnd > clientXLogPos {
				clientXLogPos = pkm.ServerWALEnd
			}
			if pkm.ReplyRequested {
				nextStatusDeadline = time.Time{}
			}

		case pglogrepl.XLogDataByteID:
			xld, _ := pglogrepl.ParseXLogData(msg.Data[1:])
			logicalMsg, _ := pglogrepl.ParseV2(xld.WALData, inStream)

			switch m := logicalMsg.(type) {
			case *pglogrepl.StreamStartMessageV2:
				inStream = true
			case *pglogrepl.StreamStopMessageV2:
				inStream = false
			case *pglogrepl.RelationMessageV2:
				relations[m.RelationID] = m
			case *pglogrepl.InsertMessageV2:
				rel := relations[m.RelationID]
				log.Printf("INSERT %s.%s", rel.Namespace, rel.RelationName)
			case *pglogrepl.UpdateMessageV2:
				rel := relations[m.RelationID]
				log.Printf("UPDATE %s.%s", rel.Namespace, rel.RelationName)
			case *pglogrepl.DeleteMessageV2:
				rel := relations[m.RelationID]
				log.Printf("DELETE %s.%s", rel.Namespace, rel.RelationName)
			}

			if xld.WALStart > clientXLogPos {
				clientXLogPos = xld.WALStart
			}
		}
	}
}
```

--------------------------------

### pglogrepl_demo Output Example

Source: https://github.com/jackc/pglogrepl/blob/master/example/pglogrepl_demo/README.md

Observe the output from the pglogrepl_demo process, which logs system information, replication slot status, and detailed WAL (Write-Ahead Logging) data for database operations.

```log
2019/08/22 20:04:35 SystemID: 6694401393180362549 Timeline: 1 XLogPos: 3/A667B740 DBName: pglogrepl
2019/08/22 20:04:35 Created temporary replication slot: pglogrepl_demo
2019/08/22 20:04:35 Logical replication started on slot pglogrepl_demo
2019/08/22 20:04:45 Sent Standby status message
2019/08/22 20:04:45 Primary Keepalive Message => ServerWALEnd: 3/A667B778 ServerTime: 2019-08-22 20:04:45.373665 -0500 CDT ReplyRequested: false
2019/08/22 20:04:51 XLogData => WALStart 3/A667B7A8 ServerWALEnd 3/A667B7A8 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData BEGIN 2435445
2019/08/22 20:04:51 XLogData => WALStart 3/A6693E30 ServerWALEnd 3/A6693E30 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData COMMIT 2435445
2019/08/22 20:04:55 Sent Standby status message
2019/08/22 20:04:55 Primary Keepalive Message => ServerWALEnd: 3/A6693E68 ServerTime: 2019-08-22 20:04:55.377208 -0500 CDT ReplyRequested: false
2019/08/22 20:04:59 XLogData => WALStart 3/A6693E68 ServerWALEnd 3/A6693E68 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData BEGIN 2435446
2019/08/22 20:04:59 XLogData => WALStart 3/A6693E68 ServerWALEnd 3/A6693E68 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData table public.t: INSERT: id[integer]:1 name[text]:'foo'
2019/08/22 20:04:59 XLogData => WALStart 3/A6693ED8 ServerWALEnd 3/A6693ED8 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData COMMIT 2435446
2019/08/22 20:05:04 XLogData => WALStart 3/A6693ED8 ServerWALEnd 3/A6693ED8 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData BEGIN 2435447
2019/08/22 20:05:04 XLogData => WALStart 3/A6693ED8 ServerWALEnd 3/A6693ED8 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData table public.t: UPDATE: id[integer]:1 name[text]:'bar'
2019/08/22 20:05:04 XLogData => WALStart 3/A6693F58 ServerWALEnd 3/A6693F58 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData COMMIT 2435447
2019/08/22 20:05:05 Sent Standby status message
2019/08/22 20:05:08 XLogData => WALStart 3/A6693F58 ServerWALEnd 3/A6693F58 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData BEGIN 2435448
2019/08/22 20:05:08 XLogData => WALStart 3/A6693F58 ServerWALEnd 3/A6693F58 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData table public.t: DELETE: (no-tuple-data)
2019/08/22 20:05:08 XLogData => WALStart 3/A6693FC0 ServerWALEnd 3/A6693FC0 ServerTime: 1999-12-31 18:00:00 -0600 CST WALData COMMIT 2435448
2019/08/22 20:05:10 Primary Keepalive Message => ServerWALEnd: 3/A6693FC0 ServerTime: 2019-08-22 20:05:10.148672 -0500 CDT ReplyRequested: false
2019/08/22 20:05:15 Sent Standby status message
2019/08/22 20:05:15 Primary Keepalive Message => ServerWALEnd: 3/A6693FF8 ServerTime: 2019-08-22 20:05:15.378933 -0500 CDT ReplyRequested: false
```

--------------------------------

### IdentifySystem - Get Server Information

Source: https://context7.com/jackc/pglogrepl/llms.txt

Demonstrates how to use the IdentifySystem function to retrieve essential information about the PostgreSQL server, such as SystemID, Timeline, and WAL position.

```APIDOC
## IdentifySystem - Get Server Information

### Description
IdentifySystem executes the IDENTIFY_SYSTEM command to retrieve information about the PostgreSQL server, including system ID, timeline, current WAL position, and database name.

### Method
N/A (Function call)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (200)
N/A (Function returns a struct)

### Code Example
```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	// Connection string must include replication=database parameter
	connString := os.Getenv("PGLOGREPL_CONN_STRING")
	// Example: "postgres://user:pass@localhost/mydb?replication=database"

	conn, err := pgconn.Connect(context.Background(), connString)
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Get system identification info
	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
	if err != nil {
		log.Fatalf("IdentifySystem failed: %v", err)
	}

	log.Printf("SystemID: %s", sysident.SystemID)
	log.Printf("Timeline: %d", sysident.Timeline)
	log.Printf("XLogPos: %s", sysident.XLogPos)
	log.Printf("DBName: %s", sysident.DBName)
}
```
```

--------------------------------

### Get PostgreSQL Server Information - Go

Source: https://context7.com/jackc/pglogrepl/llms.txt

Executes the IDENTIFY_SYSTEM command to retrieve server ID, timeline, WAL position, and database name. Requires a connection string with replication parameter and the pglogrepl and pgx packages.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	// Connection string must include replication=database parameter
	connString := os.Getenv("PGLOGREPL_CONN_STRING")
	// Example: "postgres://user:pass@localhost/mydb?replication=database"

	conn, err := pgconn.Connect(context.Background(), connString)
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Get system identification info
	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
	if err != nil {
		log.Fatalf("IdentifySystem failed: %v", err)
	}

	log.Printf("SystemID: %s", sysident.SystemID)     // Unique system identifier
	log.Printf("Timeline: %d", sysident.Timeline)     // Current timeline ID
	log.Printf("XLogPos: %s", sysident.XLogPos)       // Current WAL flush position
	log.Printf("DBName: %s", sysident.DBName)         // Connected database name
	// Output:
	// SystemID: 7123456789012345678
	// Timeline: 1
	// XLogPos: 0/1234AB8
	// DBName: mydb
}
```

--------------------------------

### Create Logical and Physical Replication Slots

Source: https://context7.com/jackc/pglogrepl/llms.txt

Demonstrates creating both logical and physical replication slots. Logical slots require an output plugin, while physical slots do not. Temporary slots are automatically removed when the connection closes.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Create a temporary logical replication slot using pgoutput plugin
	slotName := "my_replication_slot"
	outputPlugin := "pgoutput"

	result, err := pglogrepl.CreateReplicationSlot(
		context.Background(),
		conn,
		slotName,
		outputPlugin,
		pglogrepl.CreateReplicationSlotOptions{
			Temporary:      true,  // Slot removed when connection closes
				Mode:           pglogrepl.LogicalReplication,
				SnapshotAction: "EXPORT_SNAPSHOT", // Optional: EXPORT_SNAPSHOT, NOEXPORT_SNAPSHOT, USE_SNAPSHOT
		},
	)
	if err != nil {
		log.Fatalf("CreateReplicationSlot failed: %v", err)
	}

	log.Printf("SlotName: %s", result.SlotName)
	log.Printf("ConsistentPoint: %s", result.ConsistentPoint)
	log.Printf("SnapshotName: %s", result.SnapshotName)
	log.Printf("OutputPlugin: %s", result.OutputPlugin)
	// Output:
	// SlotName: my_replication_slot
	// ConsistentPoint: 0/1234AB8
	// SnapshotName: 00000003-0000001A-1
	// OutputPlugin: pgoutput

	// For physical replication, omit the output plugin and set Mode
	physResult, err := pglogrepl.CreateReplicationSlot(
		context.Background(),
		conn,
		"physical_slot",
		"",
		pglogrepl.CreateReplicationSlotOptions{
			Temporary: true,
			Mode:      pglogrepl.PhysicalReplication,
		},
	)
	if err != nil {
		log.Fatalf("CreateReplicationSlot (physical) failed: %v", err)
	}
	log.Printf("Physical slot created: %s", physResult.SlotName)
}
```

--------------------------------

### Perform Base Backups with StartBaseBackup and FinishBaseBackup

Source: https://context7.com/jackc/pglogrepl/llms.txt

These functions enable performing base backups for point-in-time recovery scenarios. They support options for progress tracking, checksums, and manifests. Requires a connection string environment variable PGLOGREPL_CONN_STRING.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Start a base backup
	result, err := pglogrepl.StartBaseBackup(
		context.Background(),
		conn,
		pglogrepl.BaseBackupOptions{
			Label:             "my_backup",
			Progress:          true,  // Enable progress reporting
			Fast:              true,  // Request fast checkpoint
			WAL:               false, // Don't include WAL in backup
			NoWait:            false, // Wait for WAL archiving
			TablespaceMap:     true,  // Include tablespace map
			NoVerifyChecksums: false, // Verify checksums
			Manifest:          true,  // Generate backup manifest (PG 13+)
			ManifestChecksums: "SHA256",
		},
	)
	if err != nil {
		log.Fatalf("StartBaseBackup failed: %v", err)
	}

	log.Printf("Backup started at LSN=%s Timeline=%d", result.LSN, result.TimelineID)
	log.Printf("Tablespaces: %d", len(result.Tablespaces))

	// Process tablespaces
	for _, ts := range result.Tablespaces {
		log.Printf("  Tablespace OID=%d Location=%s", ts.OID, ts.Location)
	}

	// Move to next tablespace/data segment
	err = pglogrepl.NextTableSpace(context.Background(), conn)
	if err != nil {
		log.Fatalf("NextTableSpace failed: %v", err)
	}

	// ... receive and save CopyData messages containing backup data ...

	// Finish the backup
	finishResult, err := pglogrepl.FinishBaseBackup(context.Background(), conn)
	if err != nil {
		log.Fatalf("FinishBaseBackup failed: %v", err)
	}

	log.Printf("Backup finished at LSN=%s Timeline=%d",
		finishResult.LSN,
		finishResult.TimelineID)
}
```

--------------------------------

### Run pglogrepl Tests with Connection String

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Execute the pglogrepl tests by setting the PGLOGREPL_TEST_CONN_STRING environment variable to a valid replication connection string.

```bash
PGLOGREPL_TEST_CONN_STRING=postgres://pglogrepl:secret@127.0.0.1/pglogrepl?replication=database go test
```

--------------------------------

### Create PostgreSQL Database

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Use this SQL command to create a dedicated database for pglogrepl testing.

```sql
create database pglogrepl;
```

--------------------------------

### PostgreSQL Replication Commands

Source: https://github.com/jackc/pglogrepl/blob/master/example/pglogrepl_demo/README.md

Execute these SQL commands in a psql session to create a table and perform insert, update, and delete operations. These actions will be logged by pglogrepl_demo.

```sql
pglogrepl@127.0.0.1:5432 pglogrepl=# create table t (id int primary key, name text);
```

```sql
pglogrepl@127.0.0.1:5432 pglogrepl=# insert into t values(1, 'foo');
```

```sql
pglogrepl@127.0.0.1:5432 pglogrepl=# update t set name='bar';
```

```sql
pglogrepl@127.0.0.1:5432 pglogrepl=# delete from t;
```

--------------------------------

### Configure pg_hba.conf for Replication

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Add this line to your pg_hba.conf file to allow the replication user to connect from localhost using MD5 authentication.

```text
host replication pglogrepl 127.0.0.1/32 md5
```

--------------------------------

### Configure postgresql.conf for Replication

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Modify your postgresql.conf file to enable logical replication by setting wal_level to 'logical' and increasing sender and slot limits.

```text
wal_level=logical
max_wal_senders=5
max_replication_slots=5
```

--------------------------------

### Parse and Use LSN - Go

Source: https://context7.com/jackc/pglogrepl/llms.txt

Demonstrates parsing LSN from PostgreSQL text format, performing arithmetic, and scanning into an LSN variable. Requires the pglogrepl package.

```go
package main

import (
	"fmt"
	"log"

	"github.com/jackc/pglogrepl"
)

func main() {
	// Parse LSN from PostgreSQL text format (XXX/XXX)
	lsn, err := pglogrepl.ParseLSN("0/1234AB8")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Parsed LSN: %s (uint64: %d)\n", lsn, uint64(lsn))
	// Output: Parsed LSN: 0/1234AB8 (uint64: 19086008)

	// LSN arithmetic for tracking progress
	newLsn := lsn + pglogrepl.LSN(1024)
	fmt.Printf("Advanced LSN: %s\n", newLsn)
	// Output: Advanced LSN: 0/1234EB8

	// LSN implements database/sql Scanner and driver.Valuer interfaces
	var scannedLsn pglogrepl.LSN
	err = scannedLsn.Scan("16/B374D848")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Scanned LSN: %s\n", scannedLsn)
	// Output: Scanned LSN: 16/B374D848
}
```

--------------------------------

### StartReplication API

Source: https://context7.com/jackc/pglogrepl/llms.txt

Initiates the streaming of changes from a replication slot. The connection enters copy-both mode after this call.

```APIDOC
## StartReplication

### Description
StartReplication begins the replication process by executing the START_REPLICATION command. After this call succeeds, the connection enters copy-both mode and begins receiving WAL data.

### Method
POST (Conceptual - this is a function call in the Go library, not a direct HTTP endpoint)

### Endpoint
N/A (Function call: `pglogrepl.StartReplication`)

### Parameters
#### Function Parameters
- **ctx** (context.Context) - Required - Context for the request.
- **conn** (*pgconn.PgConn) - Required - An active PostgreSQL connection.
- **slotName** (string) - Required - The name of the replication slot to use.
- **startLSN** (string) - Required - The LSN (Log Sequence Number) from which to start replication.
- **options** (pglogrepl.StartReplicationOptions) - Optional - Configuration options for starting replication.

##### StartReplicationOptions
- **Mode** (pglogrepl.ReplicationMode) - Required - Specifies the replication mode (`LogicalReplication` or `PhysicalReplication`).
- **PluginArgs** ([]string) - Optional - Arguments to pass to the output plugin for logical replication.

### Response
#### Success Response
No specific return value on success, typically indicated by the absence of an error. The connection will be in copy-both mode.

#### Response Example
N/A (Success is indicated by a nil error return. The connection state changes.)
```

--------------------------------

### Create PostgreSQL Replication User

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Create a PostgreSQL user with replication privileges and a password. This user will be used by pglogrepl.

```sql
create user pglogrepl with replication password 'secret';
```

--------------------------------

### Parse XLogData and PrimaryKeepaliveMessage

Source: https://context7.com/jackc/pglogrepl/llms.txt

This Go code snippet shows how to receive and parse incoming messages from a PostgreSQL logical replication stream. It differentiates between PrimaryKeepaliveMessage (heartbeats) and XLogData (WAL records) and logs relevant information. Ensure the PGLOGREPL_CONN_STRING environment variable is set.

```go
package main

import (
	"context"
	"log"
	"os"
	"time"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// ... (setup replication slot and start replication)

	clientXLogPos := pglogrepl.LSN(0) // Track position

	for {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		rawMsg, err := conn.ReceiveMessage(ctx)
		cancel()

		if err != nil {
			if pgconn.Timeout(err) {
				continue
			}
			log.Fatalf("ReceiveMessage failed: %v", err)
		}

		if errMsg, ok := rawMsg.(*pgproto3.ErrorResponse); ok {
			log.Fatalf("Received error: %+v", errMsg)
		}

		msg, ok := rawMsg.(*pgproto3.CopyData)
		if !ok {
			log.Printf("Unexpected message type: %T", rawMsg)
			continue
		}

		switch msg.Data[0] {
		case pglogrepl.PrimaryKeepaliveMessageByteID:
			// Server heartbeat message
			pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
			if err != nil {
				log.Fatalf("ParsePrimaryKeepaliveMessage failed: %v", err)
			}
			log.Printf("Keepalive: ServerWALEnd=%s ServerTime=%s ReplyRequested=%v",
				pkm.ServerWALEnd, pkm.ServerTime, pkm.ReplyRequested)

			if pkm.ServerWALEnd > clientXLogPos {
				clientXLogPos = pkm.ServerWALEnd
			}

		case pglogrepl.XLogDataByteID:
			// WAL data containing replication messages
			xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
			if err != nil {
				log.Fatalf("ParseXLogData failed: %v", err)
			}
			log.Printf("XLogData: WALStart=%s ServerWALEnd=%s ServerTime=%s DataLen=%d",
				xld.WALStart, xld.ServerWALEnd, xld.ServerTime, len(xld.WALData))

			// xld.WALData contains the logical replication message
			// Parse with pglogrepl.Parse() or pglogrepl.ParseV2()

			if xld.WALStart > clientXLogPos {
				clientXLogPos = xld.WALStart
			}
		}
	}
}
```

--------------------------------

### CreateReplicationSlot API

Source: https://context7.com/jackc/pglogrepl/llms.txt

Creates a logical or physical replication slot on the PostgreSQL server. This is a prerequisite for consuming replication data.

```APIDOC
## CreateReplicationSlot

### Description
CreateReplicationSlot creates a replication slot on the server, which is required for logical replication. The slot tracks the consumer's position and prevents WAL segments from being removed before consumption.

### Method
POST (Conceptual - this is a function call in the Go library, not a direct HTTP endpoint)

### Endpoint
N/A (Function call: `pglogrepl.CreateReplicationSlot`)

### Parameters
#### Function Parameters
- **ctx** (context.Context) - Required - Context for the request.
- **conn** (*pgconn.PgConn) - Required - An active PostgreSQL connection.
- **slotName** (string) - Required - The name for the new replication slot.
- **outputPlugin** (string) - Optional - The output plugin to use for logical replication (e.g., "pgoutput"). Omit for physical replication.
- **options** (pglogrepl.CreateReplicationSlotOptions) - Optional - Configuration options for the replication slot.

##### CreateReplicationSlotOptions
- **Temporary** (bool) - Optional - If true, the slot is removed when the connection closes.
- **Mode** (pglogrepl.ReplicationMode) - Required - Specifies the replication mode (`LogicalReplication` or `PhysicalReplication`).
- **SnapshotAction** (string) - Optional - Action to take regarding snapshots (e.g., "EXPORT_SNAPSHOT", "NOEXPORT_SNAPSHOT").

### Response
#### Success Response
- **SlotName** (string) - The name of the created replication slot.
- **ConsistentPoint** (string) - The consistent point in the WAL.
- **SnapshotName** (string) - The name of the snapshot, if created.
- **OutputPlugin** (string) - The output plugin used.

#### Response Example
```json
{
  "SlotName": "my_replication_slot",
  "ConsistentPoint": "0/1234AB8",
  "SnapshotName": "00000003-0000001A-1",
  "OutputPlugin": "pgoutput"
}
```
```

--------------------------------

### Skip Base Backup in pglogrepl Tests

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

Set the PGLOGREPL_SKIP_BASE_BACKUP environment variable to 'true' to disable the base backup during pglogrepl tests.

```bash
PGLOGREPL_SKIP_BASE_BACKUP=true
```

--------------------------------

### End Replication Session with SendStandbyCopyDone

Source: https://context7.com/jackc/pglogrepl/llms.txt

Use SendStandbyCopyDone to gracefully end copy-both mode and the replication session. It returns the final timeline and LSN position.

```go
package main

import (
	"context"
	"log"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func endReplication(conn *pgconn.PgConn) {
	// Send CopyDone to end copy-both mode
	result, err := pglogrepl.SendStandbyCopyDone(context.Background(), conn)
	if err != nil {
		log.Fatalf("SendStandbyCopyDone failed: %v", err)
	}

	if result != nil {
		log.Printf("Replication ended at Timeline=%d LSN=%s",
			result.Timeline,
			result.LSN)
	} else {
		log.Println("Replication ended (no position info)")
	}

	// Connection is now back to normal mode
	// Can execute regular queries or start replication again
}
```

--------------------------------

### Process WAL Data with pglogrepl.Parse

Source: https://context7.com/jackc/pglogrepl/llms.txt

This Go function processes WAL data by parsing it into logical replication messages. It uses a switch statement to handle different message types, logging relevant information for each. Relation messages are cached for subsequent row data decoding.

```go
package main

import (
	"log"

	"github.com/jackc/pglogrepl"
)

func processWALData(walData []byte, relations map[uint32]*pglogrepl.RelationMessage) {
	msg, err := pglogrepl.Parse(walData)
	if err != nil {
		log.Fatalf("Parse failed: %v", err)
	}

	sswitch msg := msg.(type) {
	case *pglogrepl.BeginMessage:
		log.Printf("BEGIN: FinalLSN=%s CommitTime=%s Xid=%d",
			msg.FinalLSN, msg.CommitTime, msg.Xid)

	case *pglogrepl.CommitMessage:
		log.Printf("COMMIT: CommitLSN=%s TransactionEndLSN=%s CommitTime=%s",
			msg.CommitLSN, msg.TransactionEndLSN, msg.CommitTime)

	case *pglogrepl.RelationMessage:
		// Cache relation metadata for decoding row data
		relations[msg.RelationID] = msg
		log.Printf("RELATION: ID=%d Schema=%s Table=%s Columns=%d",
			msg.RelationID, msg.Namespace, msg.RelationName, msg.ColumnNum)
		for _, col := range msg.Columns {
			log.Printf("  Column: %s (type OID: %d)", col.Name, col.DataType)
		}

	case *pglogrepl.InsertMessage:
		rel := relations[msg.RelationID]
		log.Printf("INSERT into %s.%s:", rel.Namespace, rel.RelationName)
		for i, col := range msg.Tuple.Columns {
			colName := rel.Columns[i].Name
			switch col.DataType {
			case pglogrepl.TupleDataTypeNull:
				log.Printf("  %s = NULL", colName)
			case pglogrepl.TupleDataTypeText:
				log.Printf("  %s = %s", colName, string(col.Data))
			case pglogrepl.TupleDataTypeToast:
				log.Printf("  %s = (unchanged TOAST)", colName)
			}
		}

	case *pglogrepl.UpdateMessage:
		rel := relations[msg.RelationID]
		log.Printf("UPDATE %s.%s:", rel.Namespace, rel.RelationName)
		if msg.OldTuple != nil {
			log.Printf("  OLD tuple available (type: %c)", msg.OldTupleType)
		}
		log.Printf("  NEW values: %d columns", msg.NewTuple.ColumnNum)

	case *pglogrepl.DeleteMessage:
		rel := relations[msg.RelationID]
		log.Printf("DELETE from %s.%s:", rel.Namespace, rel.RelationName)
		log.Printf("  Key/Old tuple type: %c", msg.OldTupleType)

	case *pglogrepl.TruncateMessage:
		log.Printf("TRUNCATE: %d relations, Option=%d", msg.RelationNum, msg.Option)
		for _, relID := range msg.RelationIDs {
			rel := relations[relID]
			log.Printf("  Truncated: %s.%s", rel.Namespace, rel.RelationName)
		}

	case *pglogrepl.TypeMessage:
		log.Printf("TYPE: OID=%d Schema=%s Name=%s",
			msg.DataType, msg.Namespace, msg.Name)

	case *pglogrepl.OriginMessage:
		log.Printf("ORIGIN: CommitLSN=%s Name=%s", msg.CommitLSN, msg.Name)

	case *pglogrepl.LogicalDecodingMessage:
		log.Printf("MESSAGE: LSN=%s Transactional=%v Prefix=%s Content=%s",
			msg.LSN, msg.Transactional, msg.Prefix, string(msg.Content))
	}
}

```

--------------------------------

### Retrieve Timeline History with TimelineHistory

Source: https://context7.com/jackc/pglogrepl/llms.txt

TimelineHistory retrieves the history of timeline switches for physical replication. This is useful when dealing with PostgreSQL failover and timeline changes. Requires a connection string environment variable PGLOGREPL_CONN_STRING.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	// Get history for timeline 2 (after a failover)
	timeline := int32(2)
	history, err := pglogrepl.TimelineHistory(context.Background(), conn, timeline)
	if err != nil {
		log.Fatalf("TimelineHistory failed: %v", err)
	}

	log.Printf("Timeline history file: %s", history.FileName)
	log.Printf("Content:\n%s", string(history.Content))
	// Output format:
	// 1    0/5000000    no recovery target configured
}
```

--------------------------------

### ParseV2 - Decode Protocol V2 Messages

Source: https://context7.com/jackc/pglogrepl/llms.txt

Use ParseV2 to decode logical replication messages from protocol version 2. Set inStream to true when processing messages between StreamStart and StreamStop.

```go
package main

import (
	"log"

	"github.com/jackc/pglogrepl"
)

func processV2WALData(walData []byte, relations map[uint32]*pglogrepl.RelationMessageV2, inStream *bool) {
	msg, err := pglogrepl.ParseV2(walData, *inStream)
	if err != nil {
		log.Fatalf("ParseV2 failed: %v", err)
	}

	sswitch msg := msg.(type) {
	case *pglogrepl.StreamStartMessageV2:
		// Start of a streamed transaction segment
		*inStream = true
		log.Printf("STREAM START: Xid=%d FirstSegment=%d",
			msg.Xid, msg.FirstSegment)

	case *pglogrepl.StreamStopMessageV2:
		// End of a streamed transaction segment
		*inStream = false
		log.Printf("STREAM STOP")

	case *pglogrepl.StreamCommitMessageV2:
		// Streamed transaction committed
		log.Printf("STREAM COMMIT: Xid=%d CommitLSN=%s TransactionEndLSN=%s CommitTime=%s",
			msg.Xid, msg.CommitLSN, msg.TransactionEndLSN, msg.CommitTime)

	case *pglogrepl.StreamAbortMessageV2:
		// Streamed transaction aborted
		log.Printf("STREAM ABORT: Xid=%d SubXid=%d", msg.Xid, msg.SubXid)

	case *pglogrepl.RelationMessageV2:
		relations[msg.RelationID] = msg
		log.Printf("RELATION (V2): ID=%d Schema=%s Table=%s Xid=%d",
			msg.RelationID, msg.Namespace, msg.RelationName, msg.Xid)

	case *pglogrepl.InsertMessageV2:
		rel := relations[msg.RelationID]
		log.Printf("INSERT (V2) into %s.%s (Xid=%d):",
			rel.Namespace, rel.RelationName, msg.Xid)
		for i, col := range msg.Tuple.Columns {
			if col.DataType == pglogrepl.TupleDataTypeText {
				log.Printf("  %s = %s", rel.Columns[i].Name, string(col.Data))
			}
		}

	case *pglogrepl.UpdateMessageV2:
		rel := relations[msg.RelationID]
		log.Printf("UPDATE (V2) %s.%s (Xid=%d)",
			rel.Namespace, rel.RelationName, msg.Xid)

	case *pglogrepl.DeleteMessageV2:
		rel := relations[msg.RelationID]
		log.Printf("DELETE (V2) from %s.%s (Xid=%d)",
			rel.Namespace, rel.RelationName, msg.Xid)

	case *pglogrepl.TruncateMessageV2:
		log.Printf("TRUNCATE (V2): Xid=%d Relations=%d", msg.Xid, msg.RelationNum)

	case *pglogrepl.LogicalDecodingMessageV2:
		log.Printf("MESSAGE (V2): Xid=%d Prefix=%s Content=%s",
			msg.Xid, msg.Prefix, string(msg.Content))

	// V1 messages still work in V2 mode for non-streamed transactions
	case *pglogrepl.BeginMessage:
		log.Printf("BEGIN: Xid=%d", msg.Xid)
	case *pglogrepl.CommitMessage:
		log.Printf("COMMIT: LSN=%s", msg.CommitLSN)
	}
}
```

--------------------------------

### Grant Public Schema Access (PostgreSQL 15+)

Source: https://github.com/jackc/pglogrepl/blob/master/README.md

For PostgreSQL 15 and newer, grant all privileges on the public schema to the replication user for testing purposes.

```sql
grant all on schema public to pglogrepl;
```

--------------------------------

### LSN - Log Sequence Number Type

Source: https://context7.com/jackc/pglogrepl/llms.txt

Demonstrates how to parse, manipulate, and scan PostgreSQL Log Sequence Numbers (LSN) using the pglogrepl package.

```APIDOC
## LSN - Log Sequence Number Type

### Description
The LSN type represents a PostgreSQL Log Sequence Number, used to track positions in the Write-Ahead Log. It provides parsing from PostgreSQL's text format and formatting back to string representation.

### Method
N/A (Type definition and utility functions)

### Endpoint
N/A

### Parameters
N/A

### Request Example
N/A

### Response
N/A

### Code Example
```go
package main

import (
	"fmt"
	"log"

	"github.com/jackc/pglogrepl"
)

func main() {
	// Parse LSN from PostgreSQL text format (XXX/XXX)
	lsn, err := pglogrepl.ParseLSN("0/1234AB8")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Parsed LSN: %s (uint64: %d)\n", lsn, uint64(lsn))

	// LSN arithmetic for tracking progress
	newLsn := lsn + pglogrepl.LSN(1024)
	fmt.Printf("Advanced LSN: %s\n", newLsn)

	// LSN implements database/sql Scanner and driver.Valuer interfaces
	var scannedLsn pglogrepl.LSN
	err = scannedLsn.Scan("16/B374D848")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Scanned LSN: %s\n", scannedLsn)
}
```
```

--------------------------------

### DropReplicationSlot API

Source: https://context7.com/jackc/pglogrepl/llms.txt

Removes an existing replication slot from the PostgreSQL server. Use this to clean up resources.

```APIDOC
## DropReplicationSlot

### Description
DropReplicationSlot removes an existing replication slot from the server. Use this to clean up slots that are no longer needed.

### Method
DELETE (Conceptual - this is a function call in the Go library, not a direct HTTP endpoint)

### Endpoint
N/A (Function call: `pglogrepl.DropReplicationSlot`)

### Parameters
#### Function Parameters
- **ctx** (context.Context) - Required - Context for the request.
- **conn** (*pgconn.PgConn) - Required - An active PostgreSQL connection.
- **slotName** (string) - Required - The name of the replication slot to drop.
- **options** (pglogrepl.DropReplicationSlotOptions) - Optional - Configuration options for dropping the slot.

##### DropReplicationSlotOptions
- **Wait** (bool) - Optional - If true, wait if the slot is currently active.

### Response
#### Success Response
No specific return value on success, typically indicated by the absence of an error.

#### Response Example
N/A (Success is indicated by a nil error return)
```

--------------------------------

### Drop Replication Slot

Source: https://context7.com/jackc/pglogrepl/llms.txt

Removes an existing replication slot from the server. The `Wait` option can be set to true to block until the slot is no longer active.

```go
package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
)

func main() {
	conn, err := pgconn.Connect(context.Background(), os.Getenv("PGLOGREPL_CONN_STRING"))
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close(context.Background())

	slotName := "my_replication_slot"

	err = pglogrepl.DropReplicationSlot(
		context.Background(),
		conn,
		slotName,
		pglogrepl.DropReplicationSlotOptions{
			Wait: true, // Wait if slot is active
		},
	)
	if err != nil {
		log.Fatalf("DropReplicationSlot failed: %v", err)
	}

	log.Printf("Replication slot '%s' dropped successfully", slotName)
}
```