About

• This project should stream the logical change data capture events from PostgreSQL to a file
• Should capture create, update and delete operation payload
• Should capture the existing data

Who is the end user?

• Anyone who wants to stream the change events from postgres

What is the scope of this project?

• This project captures the logical change data capture payloads from postgres and streams to the file.

How end users will use this project?

• This should be a single binary file
• End user will specify the source postgres database connection details in YML file
• End user will specify the destination disk storage where the change data capture payload files can be written

Architecture & Modules

Here is a high-level overview of how the internal modules work together to stream changes from PostgreSQL to the destination.

graph LR
    User([User]) -->|CLI Args| CLIParser[CLI Parser]

    subgraph CLIModule["CLI Module"]
        CLIParser
    end

    CLIParser -->|cli.Options| Main[Main Entry]
    Main -->|Initialize & Start| Streamer[Streamer]
    
    PostgresDB[PostgreSQL] -->|WAL Stream| Streamer

    subgraph PostgresModule["Postgres Module"]
        Streamer -->|Raw WAL Data| Parser[Parser]
        Parser -->|cdc.Event| Streamer
    end

    Streamer -->|cdc.Event| Dispatcher[Dispatcher]

    subgraph DispatcherModule["Dispatcher Module"]
        Dispatcher
    end

    Dispatcher -->|cdc.Event| SinkHandler[Sink Handler]

    subgraph EncoderModule["Encoder Module"]
        Encoder[JSONL Encoder]
    end

    subgraph SinkModule["Sink Module"]
        SinkHandler
        LocalFileSink[Local File Sink]
    end

    SinkHandler <-->|Encode| Encoder
    SinkHandler -->|cdc.EncodedEvent| LocalFileSink

    subgraph StateManagement["State Management"]
        Tracker[LSN Tracker]
    end

    LocalFileSink -->|Acknowledge LSN| Tracker
    Tracker -->|Flushed LSN| Streamer
    Streamer -.->|Standby Status Update| PostgresDB

    classDef cli fill:#E0F7FA,stroke:#00ACC1,color:#000000
    classDef main fill:#ECEFF1,stroke:#607D8B,color:#000000
    classDef postgres fill:#E3F2FD,stroke:#1E88E5,color:#000000
    classDef dispatcher fill:#FCE4EC,stroke:#D81B60,color:#000000
    classDef sink fill:#E8F5E9,stroke:#43A047,color:#000000
    classDef encoder fill:#FFFDE7,stroke:#FBC02D,color:#000000
    classDef state fill:#F3E5F5,stroke:#8E24AA,color:#000000

    class CLIParser cli
    class Main main
    class PostgresDB,Streamer,Parser postgres
    class Dispatcher dispatcher
    class SinkHandler,LocalFileSink sink
    class Encoder encoder
    class Tracker state

Loading

• CLI Module: Encapsulates command-line argument parsing and provides resolving decoupled cli.Options.
• Main (Entry Point): Initializes configuration, logger, and bootstraps the pipeline by linking the Postgres, Dispatcher, Encoder, and Sink modules together.
• Postgres Module: Manages the connection to the PostgreSQL database. It listens to logical replication slots, receives WAL (Write-Ahead Log) messages, parses them into structured cdc.Event payloads via the internal Parser, and forwards them to the Dispatcher. It also handles sending keepalive and standby status updates back to Postgres to advance the LSN (Log Sequence Number).
• Dispatcher Module: Acts as the router. It receives parsed cdc.Event payloads from the Postgres module and dispatches them to the configured downstream Sink handlers.
• Encoder Module: Provides a modular interface for transforming internal cdc.Event payloads into various storage-ready formats. The currently implemented jsonl encoder serializes events into line-delimited JSON objects, which are then passed back to the Sink Handler as cdc.EncodedEvent for persistence.
• Sink Module: The destination for CDC events. The Sink Handler coordinates the flow by using an Encoder to format the event and then writing the result to one or more Sinks. The primary implementation is localfile, which appends encoded events to local files.
• State Management: Uses an LSN Tracker to keep track of the last successfully flushed LSN. This ensures that in case of a restart, the streamer can resume from the correct position.
• CDC Module: Defines the shared domain models (like Event and EncodedEvent) that represent the data payload traversing the system.
• Config & Logger: Provide centralized configuration management and structured logging across all components.

Project Structure

.
├── docs/               # Documentation for features and architecture
├── internal/
│   ├── cdc/            # Core domain models for Change Data Capture
│   ├── cli/            # Command-line argument parsing
│   ├── config/         # Configuration management
│   ├── dispatcher/     # Event routing logic
│   ├── encoder/        # Data transformation (e.g., JSONL)
│   ├── logger/         # Structured logging
│   ├── postgres/       # Postgres streaming, parsing, and LSN tracking
│   └── sink/           # Data persistence handlers (e.g., local file)
├── go.mod              # Go module definition
├── go.sum              # Go module checksums
├── LICENSE             # Project license
├── main.go             # Application entry point
└── readme.md           # Project documentation

Getting Started

1. Build the Binary

Clone the repository and build the binary using Go:

go build -o pg-wal-stream .

2. Generate a Configuration File

Generate an example configuration file to get started:

./pg-wal-stream -init

This will create a config.example.yaml file in your current directory.

3. Update Configuration

Open config.example.yaml and update the PostgreSQL connection string, replication slot details, and the local file sink destination directory according to your environment. Make sure the destination_dir exists.

4. Run the Streamer

Start the streaming process by pointing the binary to your configuration file:

./pg-wal-stream -config config.example.yaml

(Note: You can also just pass the file as a positional argument: ./pg-wal-stream config.example.yaml)

Production Readiness & Current Limitations

This project is actively being developed. While the core logical streaming pipeline is functional, it is not yet fully production-ready. Please consider the following current limitations:

• Transaction Semantics Not Implemented: Currently, the streamer reads events directly from the WAL and streams them out immediately. It does not buffer changes or wait for a COMMIT message. As a result, events belonging to transactions that are eventually rolled back may still be streamed to your destination.
• Limited Sink Support: The system presently supports exactly 1 Sink Plugin out of the box: the localfile sink. Support for streaming data to message brokers (like Kafka, RabbitMQ, or AWS Kinesis) or other remote endpoints has not been implemented yet.