### Troubleshoot go-pq-cdc-kafka Instance Startup with Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Provides commands to diagnose why `go-pq-cdc-kafka` instances might not be starting. It involves checking container status and logs.

```bash
# Check logs
docker-compose logs go-pq-cdc-kafka

# Check health status
docker-compose ps
```

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

### Start go-pq-cdc Snapshot Process

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This bash command initiates the go-pq-cdc process with a specified configuration file. It's the primary command to start the snapshot feature, often after enabling it in the configuration.

```bash
./go-pq-cdc --config config.yml
```

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

### Scenario 1: High Throughput Test with go-pq-cdc-kafka

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Example scenario demonstrating how to set up `go-pq-cdc-kafka` for high throughput testing. It involves scaling instances, adding test data to PostgreSQL, and monitoring performance.

```bash
# Start with 5 instances
docker-compose up --scale go-pq-cdc-kafka=5 -d

# Add test data
docker-compose exec postgres psql -U cdc_user -d cdc_db -c \
  "INSERT INTO users (name) SELECT 'User' || i FROM generate_series(1, 1000000) AS i;"

# Monitor performance
docker stats
```

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

### Scenario 3: Load Testing with Varying go-pq-cdc-kafka Instance Counts

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

An example script for load testing the `go-pq-cdc-kafka` service by iterating through different numbers of instances and recording performance metrics after a stabilization period.

```bash
# Test with different instance counts
for i in 1 2 3 5 10; do
  echo "Testing with $i instances..."
  docker-compose up --scale go-pq-cdc-kafka=$i -d
  sleep 60
  # Record metrics
done
```

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

### Start Docker Compose for Benchmark

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/README.md

This command starts the necessary services defined in the docker-compose.yml file in detached mode. Ensure Docker is installed and the compose file is present in the current directory.

```sh
docker compose up -d
```

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

### Start Multiple go-pq-cdc-kafka Instances with Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

This command scales the `go-pq-cdc-kafka` service to a specified number of instances using Docker Compose. It's essential for running distributed instances of the service.

```bash
cd benchmark/benchmark_initial
docker-compose up --scale go-pq-cdc-kafka=3 -d
```

```bash
docker-compose up --scale go-pq-cdc-kafka=5 -d
```

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

### Multi-Instance Deployment Configuration (YAML)

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Configuration example for deploying multiple instances of the snapshot processor. Each instance can be configured with a unique ID and chunk size for parallel processing.

```yaml
# Instance 1
snapshot:
  instanceId: "instance-1"  # Optional: auto-generated (hostname & pid) if not set
  chunkSize: 10000

# Instance 2
snapshot:
  instanceId: "instance-2"
  chunkSize: 10000
```

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

### View Running go-pq-cdc-kafka Instances with Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Displays the running containers for the `go-pq-cdc-kafka` service. This helps in verifying the number of active instances.

```bash
docker-compose ps go-pq-cdc-kafka
```

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

### View Running go-pq-cdc-kafka Instances with Docker

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

An alternative command to view running `go-pq-cdc-kafka` containers by filtering Docker's process list. Useful for quick checks.

```bash
docker ps | grep go-pq-cdc-kafka
```

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

### Install go-pq-cdc Package

Source: https://github.com/trendyol/go-pq-cdc/blob/main/README.md

This command installs the go-pq-cdc library using the Go package manager. It fetches the latest version of the package and its dependencies, making it available for use in your Go projects.

```bash
go get github.com/Trendyol/go-pq-cdc
```

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

### Monitor Specific go-pq-cdc-kafka Instance Logs with Docker

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Allows viewing logs from a single, specific `go-pq-cdc-kafka` container instance. Useful when pinpointing issues with a particular instance.

```bash
docker logs -f benchmark_initial_go-pq-cdc-kafka_1
```

```bash
docker logs -f benchmark_initial_go-pq-cdc-kafka_2
```

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

### Scale go-pq-cdc-kafka Instances While Running with Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Allows dynamic scaling of the `go-pq-cdc-kafka` service up or down while it's already running. This is crucial for adjusting capacity based on load.

```bash
docker-compose up --scale go-pq-cdc-kafka=5 -d
```

```bash
docker-compose up --scale go-pq-cdc-kafka=2 -d
```

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

### Scenario 2: Graceful Scale Down of go-pq-cdc-kafka Instances

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Demonstrates a gradual scaling down process for `go-pq-cdc-kafka` instances using Docker Compose. This approach can help in ensuring a smoother transition during load reduction.

```bash
# First show existing instances
docker-compose ps go-pq-cdc-kafka

# Gradually scale down
docker-compose up --scale go-pq-cdc-kafka=3 -d
sleep 30
docker-compose up --scale go-pq-cdc-kafka=1 -d
```

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

### Go SQL Queries for Ordered Data Retrieval

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Provides Go code examples for retrieving data in ordered chunks using SQL. It covers ordering by primary key, composite keys, and falls back to `ctid` when no primary key is available. Primary keys are recommended for optimal performance.

```go
// Primary key ordering (preferred)
SELECT * FROM users ORDER BY id LIMIT 10000 OFFSET 0

// Composite key ordering
SELECT * FROM orders ORDER BY user_id, created_at LIMIT 10000 OFFSET 0

// No primary key (fallback to ctid)
SELECT * FROM logs ORDER BY ctid LIMIT 10000 OFFSET 0
```

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

### Complete CDC Job and Start Replication (SQL & Go)

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Checks if all chunks are completed, marks the job as finished, closes the snapshot transaction, sends an END marker, and then starts logical replication from the snapshot LSN.

```sql
// 1. Check if all chunks completed
allDone := SELECT COUNT(*) = COUNT(*) FILTER (WHERE status = 'completed')
           FROM cdc_snapshot_chunks
           WHERE slot_name = 'cdc_slot'

if allDone {
    // 2. Mark job as completed
    UPDATE cdc_snapshot_job
    SET completed = true
    WHERE slot_name = 'cdc_slot'
    
    // 3. Close snapshot transaction
    COMMIT  // (on coordinator's export connection)
    
    // 4. Send END marker
    handler(&format.Snapshot{
        EventType: format.SnapshotEventTypeEnd,
        LSN:       snapshotLSN,
    })
    
    // 5. Start CDC from snapshot LSN
    START_REPLICATION SLOT cdc_slot LOGICAL 0/12345678
    // CDC stream starts from where snapshot was taken
}
```

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

### Manually Define go-pq-cdc-kafka Instances in Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Illustrates how to define individual `go-pq-cdc-kafka` service instances manually within a `docker-compose.yml` file. This allows for distinct configurations for each instance.

```yaml
go-pq-cdc-kafka-1:
  build:
    context: ../../
    dockerfile: ./benchmark/benchmark_initial/go-pq-cdc-kafka/Dockerfile
  # ... other settings

go-pq-cdc-kafka-2:
  build:
    context: ../../
    dockerfile: ./benchmark/benchmark_initial/go-pq-cdc-kafka/Dockerfile
  # ... other settings

go-pq-cdc-kafka-3:
  build:
    context: ../../
    dockerfile: ./benchmark/benchmark_initial/go-pq-cdc-kafka/Dockerfile
  # ... other settings
```

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

### Troubleshoot go-pq-cdc-kafka Resource Usage with Docker

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Checks the real-time resource (CPU, memory) consumption of running Docker containers, including `go-pq-cdc-kafka` instances. Essential for identifying resource-related problems.

```bash
docker stats
```

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

### Monitor go-pq-cdc-kafka Instance Logs with Docker Compose

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Streams logs from all running `go-pq-cdc-kafka` instances in real-time. This is vital for debugging and monitoring service behavior.

```bash
docker-compose logs -f go-pq-cdc-kafka
```

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

### Basic go-pq-cdc Connector Usage in Go

Source: https://github.com/trendyol/go-pq-cdc/blob/main/README.md

Demonstrates the basic setup and usage of the go-pq-cdc connector in Go. It includes configuration for connecting to a PostgreSQL database, setting up replication, and defining a handler function to process incoming change data messages (inserts, deletes, updates).

```go
package main

import (
	"context"
	cdc "github.com/Trendyol/go-pq-cdc"
	"github.com/Trendyol/go-pq-cdc/config"
	"github.com/Trendyol/go-pq-cdc/pq/message/format"
	"github.com/Trendyol/go-pq-cdc/pq/publication"
	"github.com/Trendyol/go-pq-cdc/pq/replication"
	"github.com/Trendyol/go-pq-cdc/pq/slot"
	"log/slog"
	"os"
)

func main() {
	ctx := context.Background()
	cfg := config.Config{
		Host:      "127.0.0.1",
		Username:  "cdc_user",
		Password:  "cdc_pass",
		Database:  "cdc_db",
		DebugMode: false,
		Publication: publication.Config{
			CreateIfNotExists: true,
			Name: "cdc_publication",
			Operations: publication.Operations{
				publication.OperationInsert,
				publication.OperationDelete,
				publication.OperationTruncate,
				publication.OperationUpdate,
			},
			Tables: publication.Tables{publication.Table{
				Name:            "users",
				ReplicaIdentity: publication.ReplicaIdentityFull,
			}},
		},
		Slot: slot.Config{
			Name:                        "cdc_slot",
			CreateIfNotExists:           true,
			SlotActivityCheckerInterval: 3000,
		},
		Metric: config.MetricConfig{
			Port: 8081,
		},
		Logger: config.LoggerConfig{
			LogLevel: slog.LevelInfo,
		},
	}

	connector, err := cdc.NewConnector(ctx, cfg, Handler)
	if err != nil {
		slog.Error("new connector", "error", err)
		os.Exit(1)
	}

	defer connector.Close()
	connector.Start(ctx)
}

func Handler(ctx *replication.ListenerContext) {
	switch msg := ctx.Message.(type) {
	case *format.Insert:
		slog.Info("insert message received", "new", msg.Decoded)
	case *format.Delete:
		slog.Info("delete message received", "old", msg.OldDecoded)
	case *format.Update:
		slog.Info("update message received", "new", msg.NewDecoded, "old", msg.OldDecoded)
	}

	if err := ctx.Ack(); err != nil {
		slog.Error("ack", "error", err)
	}
}

```

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

### Configure go-pq-cdc Snapshot to Include Specific Tables

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This YAML configuration illustrates how to specify which tables should be included in the go-pq-cdc snapshot. Only tables listed under `publication.tables` will be snapshotted. Tables not listed, like 'products' in the example, will be excluded.

```yaml
publication:
  tables:
    - name: users      # ✅ Included in snapshot
    - name: orders     # ✅ Included in snapshot
    # products table NOT listed → NOT in snapshot
```

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

### Test go-pq-cdc Snapshot Disaster Recovery

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This bash script outlines a process for testing disaster recovery scenarios for the go-pq-cdc snapshot feature. It involves starting a snapshot, killing it mid-process, verifying the intermediate state, and then restarting to ensure recovery.

```bash
# 1. Start snapshot
./go-pq-cdc --config config.yml &
PID=$!

# 2. Kill mid-snapshot
sleep 10 && kill -9 $PID

# 3. Verify state
psql -c "SELECT * FROM cdc_snapshot_chunks WHERE status = 'in_progress';"

# 4. Restart and verify recovery
./go-pq-cdc --config config.yml
```

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

### Resume Snapshot on Failure Example

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Illustrates the resume capability of the snapshot_only mode. If a snapshot process is interrupted, restarting the process automatically continues from where it left off, ensuring no data is missed.

```bash
# Run 1: Process 500/1000 chunks → CRASH
# Run 2: Automatically continues from chunk 501 ✅
```

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

### Troubleshoot Prometheus Discovery of go-pq-cdc-kafka Instances

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

A command to test DNS resolution from within the Prometheus container, helping to diagnose why Prometheus might not be discovering the `go-pq-cdc-kafka` instances.

```bash
docker-compose exec prometheus nslookup tasks.go-pq-cdc-kafka
```

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

### Create Basic CDC Connector in Go

Source: https://context7.com/trendyol/go-pq-cdc/llms.txt

Initializes a Go connector to capture real-time changes from PostgreSQL tables. This example uses inline configuration for database connection, publication, and slot details.

```go
package main

import (
    "context"
    "log/slog"
    "os"

    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/publication"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
    "github.com/Trendyol/go-pq-cdc/pq/slot"
)

func main() {
    ctx := context.Background()

    cfg := config.Config{
        Host:     "127.0.0.1",
        Port:     5432,
        Username: "cdc_user",
        Password: "cdc_pass",
        Database: "cdc_db",
        Publication: publication.Config{
            CreateIfNotExists: true,
            Name: "cdc_publication",
            Operations: publication.Operations{
                publication.OperationInsert,
                publication.OperationUpdate,
                publication.OperationDelete,
                publication.OperationTruncate,
            },
            Tables: publication.Tables{
                {
                    Name:            "users",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
            },
        },
        Slot: slot.Config{
            Name:                        "cdc_slot",
            CreateIfNotExists:           true,
            SlotActivityCheckerInterval: 3000,
        },
        Metric: config.MetricConfig{
            Port: 8081,
        },
        Logger: config.LoggerConfig{
            LogLevel: slog.LevelInfo,
        },
    }

    connector, err := cdc.NewConnector(ctx, cfg, handleMessage)
    if err != nil {
        slog.Error("connector initialization failed", "error", err)
        os.Exit(1)
    }
    defer connector.Close()

    connector.Start(ctx)
}

func handleMessage(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    case *format.Insert:
        slog.Info("insert", "table", "users", "data", msg.Decoded)
    case *format.Update:
        slog.Info("update", "old", msg.OldDecoded, "new", msg.NewDecoded)
    case *format.Delete:
        slog.Info("delete", "data", msg.OldDecoded)
    }

    if err := ctx.Ack(); err != nil {
        slog.Error("ack failed", "error", err)
    }
}
```

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

### PostgreSQL WAL Configuration for CDC

Source: https://github.com/trendyol/go-pq-cdc/blob/main/benchmark/benchmark_initial/SCALING_GUIDE.md

Specifies the necessary PostgreSQL Write-Ahead Log (WAL) settings required for logical replication, which is fundamental for Change Data Capture (CDC) operations.

```sql
wal_level=logical
max_wal_senders=10
max_replication_slots=10
```

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

### Go: Configure Multi-Table CDC with Diverse Replica Identities

Source: https://context7.com/trendyol/go-pq-cdc/llms.txt

This Go code snippet illustrates the setup of a CDC connector using go-pq-cdc. It configures a publication to monitor several tables, each with potentially different replica identity settings. The code defines the connection parameters, publication details (including table names, schemas, and replica identities), slot configuration, and logging. It also includes a message routing function to handle different types of CDC events (INSERT, UPDATE, DELETE) and acknowledges them.

```go
package main

import (
    "context"
    "encoding/json"
    "log/slog"
    "os"

    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/publication"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
    "github.com/Trendyol/go-pq-cdc/pq/slot"
)

func main() {
    ctx := context.Background()

    cfg := config.Config{
        Host:     "127.0.0.1",
        Port:     5432,
        Username: "cdc_user",
        Password: "cdc_pass",
        Database: "cdc_db",
        Publication: publication.Config{
            CreateIfNotExists: true,
            Name: "multi_table_publication",
            Operations: publication.Operations{
                publication.OperationInsert,
                publication.OperationUpdate,
                publication.OperationDelete,
            },
            Tables: publication.Tables{
                {
                    Name:            "users",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
                {
                    Name:            "orders",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
                {
                    Name:            "products",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityDefault,
                },
                {
                    Name:            "audit_logs",
                    Schema:          "audit",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
            },
        },
        Slot: slot.Config{
            Name:                        "multi_table_slot",
            CreateIfNotExists:           true,
            SlotActivityCheckerInterval: 3000,
        },
        Metric: config.MetricConfig{
            Port: 8081,
        },
        Logger: config.LoggerConfig{
            LogLevel: slog.LevelInfo,
        },
    }

    connector, err := cdc.NewConnector(ctx, cfg, routeMessage)
    if err != nil {
        slog.Error("connector failed", "error", err)
        os.Exit(1)
    }
    defer connector.Close()

    connector.Start(ctx)
}

func routeMessage(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    case *format.Insert:
        handleInsert(msg)
    case *format.Update:
        handleUpdate(msg)
    case *format.Delete:
        handleDelete(msg)
    }

    if err := ctx.Ack(); err != nil {
        slog.Error("acknowledgment failed", "error", err)
    }
}

func handleInsert(msg *format.Insert) {
    data, _ := json.Marshal(msg.Decoded)
    slog.Info("insert event",
        "table", msg.Namespace + "." + msg.Relation,
        "data", string(data))
}

func handleUpdate(msg *format.Update) {
    newData, _ := json.Marshal(msg.NewDecoded)
    oldData, _ := json.Marshal(msg.OldDecoded)
    slog.Info("update event",
        "table", msg.Namespace + "." + msg.Relation,
        "old", string(oldData),
        "new", string(newData))
}

func handleDelete(msg *format.Delete) {
    data, _ := json.Marshal(msg.OldDecoded)
    slog.Info("delete event",
        "table", msg.Namespace + "." + msg.Relation,
        "data", string(data))
}

```

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

### Go Snapshot Mode with Initial Data Capture using go-pq-cdc

Source: https://context7.com/trendyol/go-pq-cdc/llms.txt

This Go program configures and runs the go-pq-cdc library in snapshot mode. It captures existing data from specified tables before starting Change Data Capture (CDC). The configuration includes connection details, publication settings, slot configuration, and snapshot parameters. The handler function processes snapshot events and CDC messages.

```go
package main

import (
    "context"
    "encoding/json"
    "log"
    "log/slog"
    "os"
    "time"

    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/publication"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
    "github.com/Trendyol/go-pq-cdc/pq/slot"
)

func main() {
    ctx := context.Background()

    cfg := config.Config{
        Host:     "127.0.0.1",
        Port:     5432,
        Username: "cdc_user",
        Password: "cdc_pass",
        Database: "cdc_db",
        Publication: publication.Config{
            CreateIfNotExists: true,
            Name: "cdc_publication",
            Operations: publication.Operations{
                publication.OperationInsert,
                publication.OperationUpdate,
                publication.OperationDelete,
            },
            Tables: publication.Tables{
                {
                    Name:            "users",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
                {
                    Name:            "orders",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
            },
        },
        Slot: slot.Config{
            Name:              "cdc_slot",
            CreateIfNotExists: true,
        },
        Snapshot: config.SnapshotConfig{
            Enabled:           true,
            Mode:              config.SnapshotModeInitial,
            ChunkSize:         10000,
            ClaimTimeout:      30 * time.Second,
            HeartbeatInterval: 5 * time.Second,
        },
        Metric: config.MetricConfig{
            Port: 8081,
        },
        Logger: config.LoggerConfig{
            LogLevel: slog.LevelInfo,
        },
    }

    connector, err := cdc.NewConnector(ctx, cfg, handleSnapshotAndCDC)
    if err != nil {
        slog.Error("connector error", "error", err)
        os.Exit(1)
    }
    defer connector.Close()

    connector.Start(ctx)
}

func handleSnapshotAndCDC(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    case *format.Snapshot:
        handleSnapshot(msg)
    case *format.Insert:
        log.Printf("CDC INSERT: table=%s, data=%v", "users", msg.Decoded)
    case *format.Update:
        log.Printf("CDC UPDATE: old=%v, new=%v", msg.OldDecoded, msg.NewDecoded)
    case *format.Delete:
        log.Printf("CDC DELETE: data=%v", msg.OldDecoded)
    }

    if err := ctx.Ack(); err != nil {
        log.Printf("ACK error: %v", err)
    }
}

func handleSnapshot(s *format.Snapshot) {
    switch s.EventType {
    case format.SnapshotEventTypeBegin:
        log.Printf("📸 Snapshot BEGIN | LSN: %s | Time: %s",
            s.LSN.String(),
            s.ServerTime.Format("2006-01-02 15:04:05"))

    case format.SnapshotEventTypeData:
        data, _ := json.MarshalIndent(s.Data, "", "  ")
        log.Printf("📸 Snapshot DATA | Table: %s.%s | Row: %s",
            s.Schema,
            s.Table,
            string(data))

    case format.SnapshotEventTypeEnd:
        log.Printf("📸 Snapshot END | LSN: %s | Total Duration: complete",
            s.LSN.String())
    }
}

```

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

### Go CDC Connector with Snapshot

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This Go code demonstrates how to initialize and run a Change Data Capture (CDC) connector using the go-pq-cdc library. It includes loading configuration from a YAML file, setting up a message handler for both snapshot and CDC events (INSERT, UPDATE, DELETE), and managing the connector's lifecycle. The handler processes snapshot events (BEGIN, DATA, END) and real-time data changes, acknowledging each message. Dependencies include the go-pq-cdc library and its related packages.

```go
package main

import (
    "context"
    "log"
    "encoding/json"
    
    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
)

func main() {
    ctx := context.Background()
    
    // Load configuration
    cfg, err := config.ReadConfigYAML("config.yml")
    if err != nil {
        log.Fatal(err)
    }
    
    // Create connector with snapshot support
    connector, err := cdc.NewConnector(ctx, cfg, handleMessage)
    if err != nil {
        log.Fatal(err)
    }
    defer connector.Close()
    
    // Start (snapshot + CDC)
    connector.Start(ctx)
}

func handleMessage(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    
    // Snapshot events
    case *format.Snapshot:
        handleSnapshot(msg)
        
    // CDC events
    case *format.Insert:
        log.Printf("INSERT: %v", msg.Decoded)
    case *format.Update:
        log.Printf("UPDATE: %v -> %v", msg.OldDecoded, msg.NewDecoded)
    case *format.Delete:
        log.Printf("DELETE: %v", msg.OldDecoded)
    }
    
    if err := ctx.Ack(); err != nil {
        log.Printf("ACK error: %v", err)
    }
}

func handleSnapshot(s *format.Snapshot) {
    switch s.EventType {
    case format.SnapshotEventTypeBegin:
        log.Printf("📸 Snapshot BEGIN | LSN: %s", s.LSN.String())
        
    case format.SnapshotEventTypeData:
        data, _ := json.Marshal(s.Data)
        log.Printf("📸 Snapshot DATA | Table: %s.%s | Data: %s",
            s.Schema, s.Table, string(data))
        
    case format.SnapshotEventTypeEnd:
        log.Printf("📸 Snapshot END | LSN: %s", s.LSN.String())
    }
}

```

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

### Bash Command to Deploy Multiple Instances

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This bash command shows how to launch multiple instances of the application in parallel using background processes. This is useful for distributing the workload and speeding up snapshotting.

```bash
# Deploy more instances for parallel processing
./app --config config1.yml & 
./app --config config2.yml & 
./app --config config3.yml & 
```

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

### YAML Configuration for Heartbeat and Timeout Tuning

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Illustrates tuning heartbeat intervals and claim timeouts in YAML for failure detection. Aggressive settings detect failures faster but increase database load, while conservative settings reduce load but slow down recovery.

```yaml
# Fast failure detection (aggressive)
snapshot:
  heartbeatInterval: 5s
  claimTimeout: 15s

# Slow failure detection (conservative)
snapshot:
  heartbeatInterval: 30s
  claimTimeout: 120s
```

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

### Configure Initial Snapshot Mode in go-pq-cdc Production

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This configuration snippet demonstrates how to set the snapshot mode to 'initial' in the go-pq-cdc configuration file. This is a recommended best practice for production environments to ensure snapshots are taken only once, preventing re-snapshotting on every restart.

```yaml
snapshot:
  mode: initial  # ✅ Safe: takes snapshot only once
```

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

### Configure go-pq-cdc Snapshot with TimescaleDB Hypertables

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This YAML configuration demonstrates how to enable and configure the snapshot feature in go-pq-cdc when using TimescaleDB hypertables. It shows how to include hypertables in the publication and set a suitable `chunkSize`, highlighting that the system treats hypertables as regular tables during snapshotting.

```yaml
publication:
  tables:
    - name: metrics  # Hypertable
      schema: public
      replicaIdentity: FULL

snapshot:
  enabled: true
  chunkSize: 10000  # Works with TimescaleDB chunks internally
```

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

### YAML Configuration for Snapshot Chunk Size

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Demonstrates how to configure the snapshot chunk size in YAML based on table size. Smaller chunks offer more parallelism but increase overhead, while larger chunks reduce overhead at the cost of parallelism.

```yaml
# Small tables (< 100K rows)
snapshot:
  chunkSize: 5000  # Smaller chunks, more overhead but fine-grained

# Medium tables (100K - 1M rows)
snapshot:
  chunkSize: 10000  # Balanced

# Large tables (> 1M rows)
snapshot:
  chunkSize: 50000  # Larger chunks, less overhead
```

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

### Snapshot Configuration in YAML

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This YAML snippet illustrates the configuration for the snapshot feature within the go-pq-cdc connector. It details parameters such as enabling the snapshot (`enabled: true`), the snapshot mode (`mode: initial`), the number of rows processed per chunk (`chunkSize`), and timeouts for claiming chunks and worker heartbeats. These settings control how historical data is captured before switching to real-time CDC.

```yaml
snapshot:
  enabled: true                # Enable snapshot feature
  mode: initial                # Take snapshot only if not completed before
  chunkSize: 10000             # Rows per chunk
  claimTimeout: 30s            # Reclaim stale chunks after 30s
  heartbeatInterval: 10s       # Send heartbeat every 10s

```

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

### GET /metrics - Exposed Metrics

Source: https://github.com/trendyol/go-pq-cdc/blob/main/README.md

This endpoint exposes various metrics related to PostgreSQL change data capture (CDC) operations. These metrics provide insights into update, delete, and insert operations, as well as latency and replication slot status.

```APIDOC
## GET /metrics

### Description
Exposes metrics related to PostgreSQL change data capture (CDC) operations.

### Method
GET

### Endpoint
/metrics

### Parameters

#### Query Parameters
*None*

#### Request Body
*None*

### Request Example
*No request body*

### Response
#### Success Response (200)
- **Metric Name** (string) - The name of the collected metric.
- **Description** (string) - A description of what the metric represents.
- **Labels** (array of strings) - Labels associated with the metric (e.g., slot_name, host).
- **Value Type** (string) - The type of the metric value (e.g., Counter, Gauge).

#### Response Example
```json
{
  "go_pq_cdc_update_total": {
    "description": "The total number of `UPDATE` operations captured on specific tables.",
    "labels": ["slot_name", "host"],
    "value_type": "Counter"
  },
  "go_pq_cdc_delete_total": {
    "description": "The total number of `DELETE` operations captured on specific tables.",
    "labels": ["slot_name", "host"],
    "value_type": "Counter"
  },
  "go_pq_cdc_insert_total": {
    "description": "The total number of `INSERT` operations captured on specific tables.",
    "labels": ["slot_name", "host"],
    "value_type": "Counter"
  },
  "go_pq_cdc_cdc_latency_current": {
    "description": "The current latency in capturing data changes from PostgreSQL.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_process_latency_current": {
    "description": "The current latency in processing the captured data changes.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_replication_slot_slot_confirmed_flush_lsn": {
    "description": "The last confirmed flush Log Sequence Number (LSN) in the PostgreSQL replication slot.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_replication_slot_slot_current_lsn": {
    "description": "The current Log Sequence Number (LSN) being processed in the PostgreSQL replication slot.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_replication_slot_slot_is_active": {
    "description": "Indicates whether the PostgreSQL replication slot is currently active (1 for active, 0 for inactive).",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_replication_slot_slot_lag": {
    "description": "The replication lag measured by the difference between the current LSN and the confirmed flush LSN.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_replication_slot_slot_retained_wal_size": {
    "description": "The size of Write-Ahead Logging (WAL) files retained for the replication slot in bytes.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_snapshot_in_progress": {
    "description": "Indicates whether snapshot is currently in progress (1 for active, 0 for inactive).",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_snapshot_total_tables": {
    "description": "Total number of tables to snapshot.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_snapshot_total_chunks": {
    "description": "Total number of chunks to process across all tables.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_snapshot_completed_chunks": {
    "description": "Number of chunks completed in snapshot.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  },
  "go_pq_cdc_snapshot_total_rows": {
    "description": "Total number of rows read during snapshot.",
    "labels": ["slot_name", "host"],
    "value_type": "Counter"
  },
  "go_pq_cdc_snapshot_duration_seconds": {
    "description": "Duration of the last snapshot operation in seconds.",
    "labels": ["slot_name", "host"],
    "value_type": "Gauge"
  }
}
```

### Error Handling
*No specific error responses documented for this endpoint.*
```

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

### Create cdc_snapshot_job Table SQL

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

SQL statement to create the 'cdc_snapshot_job' table. This table stores metadata for snapshot jobs, including slot name, snapshot ID, LSN, start time, completion status, and chunk counts. It is essential for managing the lifecycle of snapshot operations in the CDC process.

```sql
CREATE TABLE cdc_snapshot_job (
    slot_name        TEXT PRIMARY KEY,
    snapshot_id      TEXT NOT NULL,
    snapshot_lsn     TEXT NOT NULL,
    started_at       TIMESTAMP NOT NULL,
    completed        BOOLEAN DEFAULT FALSE,
    total_chunks     INT NOT NULL DEFAULT 0,
    completed_chunks INT NOT NULL DEFAULT 0
);
```

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

### Configure Heartbeat Query for PostgreSQL CDC

Source: https://github.com/trendyol/go-pq-cdc/blob/main/README.md

Specifies the SQL statement executed on the source PostgreSQL database for each heartbeat. This statement should trigger Write-Ahead Logging (WAL), such as an INSERT into a dedicated heartbeat table included in the publication. An example is provided for inserting into a 'heartbeat_events' table.

```sql
INSERT INTO public.heartbeat_events(txt) VALUES ('hb')
```

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

### Multi-Pod Snapshot Only Deployment Configuration

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

Demonstrates how to configure multiple pods to run snapshot_only mode in parallel for faster data snapshotting. The configuration is identical across pods, relying on an internal election mechanism for coordination.

```yaml
# Same config on all pods
snapshot:
  enabled: true
  mode: snapshot_only
  chunkSize: 10000

# All pods will:
# - Use consistent slot name: "snapshot_only_<database>"
# - Participate in coordinator election (one becomes coordinator)
# - Process chunks in parallel (work together)
# - Resume if any pod crashes (no duplicate work)
```

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

### Enable TimescaleDB Hypertable CDC in Go

Source: https://context7.com/trendyol/go-pq-cdc/llms.txt

This Go code snippet demonstrates how to configure and start a CDC connector for TimescaleDB hypertables. It includes setting up publication, replication slot, and extension support for TimescaleDB. The handler function processes insert, update, and delete messages from the hypertable.

```go
package main

import (
    "context"
    "log/slog"
    "os"

    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/publication"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
    "github.com/Trendyol/go-pq-cdc/pq/slot"
)

func main() {
    ctx := context.Background()

    cfg := config.Config{
        Host:     "127.0.0.1",
        Port:     5432,
        Username: "cdc_user",
        Password: "cdc_pass",
        Database: "timescale_db",
        Publication: publication.Config{
            CreateIfNotExists: true,
            Name: "timescale_publication",
            Operations: publication.Operations{
                publication.OperationInsert,
                publication.OperationUpdate,
                publication.OperationDelete,
            },
            Tables: publication.Tables{
                {
                    Name:            "metrics",
                    Schema:          "public",
                    ReplicaIdentity: publication.ReplicaIdentityFull,
                },
            },
        },
        Slot: slot.Config{
            Name:              "timescale_slot",
            CreateIfNotExists: true,
        },
        ExtensionSupport: config.ExtensionSupport{
            EnableTimeScaleDB: true,
        },
        Metric: config.MetricConfig{
            Port: 8081,
        },
        Logger: config.LoggerConfig{
            LogLevel: slog.LevelInfo,
        },
    }

    connector, err := cdc.NewConnector(ctx, cfg, handleTimescaleData)
    if err != nil {
        slog.Error("connector error", "error", err)
        os.Exit(1)
    }
    defer connector.Close()

    slog.Info("Starting CDC for TimescaleDB hypertables")
    connector.Start(ctx)
}

func handleTimescaleData(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    case *format.Insert:
        slog.Info("timescale insert",
            "table", msg.Relation,
            "data", msg.Decoded)
    case *format.Update:
        slog.Info("timescale update",
            "table", msg.Relation,
            "old", msg.OldDecoded,
            "new", msg.NewDecoded)
    case *format.Delete:
        slog.Info("timescale delete",
            "table", msg.Relation,
            "data", msg.OldDecoded)
    }

    if err := ctx.Ack(); err != nil {
        slog.Error("ack error", "error", err)
    }
}

```

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

### Snapshot Mode 'never' Configuration

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This YAML configuration sets the snapshot mode to 'never'. When this mode is active, the go-pq-cdc connector will completely skip the snapshot phase. It will immediately begin capturing changes from the current position in the transaction log. This mode is suitable for scenarios where historical data capture is not required, and the CDC process can start from the most recent available data.

```yaml
snapshot:
  mode: never

```

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

### Minimal Configuration for Snapshot Only

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

A minimal configuration for snapshot_only mode, highlighting that publication and replication slots are not required and can be omitted. Includes basic connection details and optional metric port.

```yaml
host: localhost
database: mydb
username: user
password: pass

# Publication and slot NOT required for snapshot_only
# Can be omitted or left empty

publication:
  tables:
    - name: users
      schema: public
    - name: orders
      schema: public

snapshot:
  enabled: true
  mode: snapshot_only
  chunkSize: 10000

metric:
  port: 8081
```

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

### Loading Configuration from YAML File

Source: https://context7.com/trendyol/go-pq-cdc/llms.txt

Load configuration from an external YAML file for environment-specific settings.

```APIDOC
## Loading Configuration from YAML File

### Description
Load configuration from external YAML file for environment-specific settings.

### Method
N/A (Code Example)

### Endpoint
N/A (Code Example)

### Parameters
N/A (Configuration is done via file)

### Request Example
```go
package main

import (
    "context"
    "log/slog"
    "os"

    cdc "github.com/Trendyol/go-pq-cdc"
    "github.com/Trendyol/go-pq-cdc/config"
    "github.com/Trendyol/go-pq-cdc/pq/message/format"
    "github.com/Trendyol/go-pq-cdc/pq/replication"
)

func main() {
    ctx := context.Background()

    cfg, err := config.ReadConfigYAML("config.yml")
    if err != nil {
        slog.Error("failed to read config", "error", err)
        os.Exit(1)
    }

    connector, err := cdc.NewConnectorWithConfigFile(ctx, "config.yml", handler)
    if err != nil {
        slog.Error("connector creation failed", "error", err)
        os.Exit(1)
    }
    defer connector.Close()

    connector.Start(ctx)
}

func handler(ctx *replication.ListenerContext) {
    switch msg := ctx.Message.(type) {
    case *format.Insert:
        slog.Info("insert detected", "data", msg.Decoded)
    case *format.Update:
        slog.Info("update detected", "new", msg.NewDecoded)
    case *format.Delete:
        slog.Info("delete detected", "old", msg.OldDecoded)
    }

    ctx.Ack()
}
```
```

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

### Configuration File for go-pq-cdc

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This YAML file configures the PostgreSQL Change Data Capture (CDC) connector. It specifies database connection details, publication settings (including operations and tables to monitor), replication slot parameters, and snapshot behavior (enabled, mode, chunk size, timeouts). It also includes settings for logging and metrics. The `snapshot.mode: initial` is recommended for production use cases.

```yaml
host: localhost
database: mydb
port: 5432
username: cdc_user
password: cdc_pass

publication:
  name: cdc_publication
  createIfNotExists: true
  operations:
    - INSERT
    - UPDATE
    - DELETE
  tables:
    - name: users
      schema: public
      replicaIdentity: FULL
    - name: orders
      schema: public
      replicaIdentity: FULL

slot:
  name: cdc_slot
  createIfNotExists: true
  slotActivityCheckerInterval: 3000

# Snapshot Configuration
snapshot:
  enabled: true                # Enable snapshot feature
  mode: initial                # Take snapshot only if not completed before
  chunkSize: 10000             # Rows per chunk
  claimTimeout: 30s            # Reclaim stale chunks after 30s
  heartbeatInterval: 10s       # Send heartbeat every 10s

logger:
  logLevel: info

metric:
  port: 8081

```

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

### Create CDC Job and Chunk Metadata (Go/SQL)

Source: https://github.com/trendyol/go-pq-cdc/blob/main/docs/SNAPSHOT_FEATURE.md

This snippet demonstrates the Go code and SQL statements used to create job and chunk metadata for a CDC process. It first captures the current LSN, then inserts job details into `cdc_snapshot_job`, and finally splits tables into manageable chunks by inserting records into `cdc_snapshot_chunks`. This approach prevents Out-of-Memory errors by processing data in batches.

```go
// 1. Capture current LSN
currentLSN := SELECT pg_current_wal_lsn()  // e.g., "0/12345678"

// 2. Create job metadata
INSERT INTO cdc_snapshot_job (
    slot_name, snapshot_id, snapshot_lsn, started_at,
    completed, total_chunks, completed_chunks
) VALUES (
    'cdc_slot', 'PENDING', '0/12345678', NOW(),
    false, 0, 0
)

// 3. Split tables into chunks
for each table in publication.tables {
    rowCount := SELECT COUNT(*) FROM schema.table
    numChunks := ceiling(rowCount / chunkSize)
    
    for i := 0 to numChunks-1 {
        INSERT INTO cdc_snapshot_chunks (
            slot_name, table_schema, table_name,
            chunk_index, chunk_start, chunk_size, status
        ) VALUES (
            'cdc_slot', 'public', 'users',
            i, i * chunkSize, chunkSize, 'pending'
        )
    }
}
```