### Start Example Project Development Server

Source: https://github.com/peerdb-io/peerdb/blob/main/ui/README.md

Starts the Next.js development server for the example project. Access the running application at http://localhost:3000.

```bash
npm run dev
```

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

### SQL Interface Examples

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/README.md

Examples of SQL commands for creating peers and mirrors, and querying mirror status using the Nexus Server.

```sql
-- Create peer
CREATE PEER pg_dest AS postgres_config(...);

-- Start CDC
CREATE MIRROR my_mirror AS CDC FROM pg_source TO sf_dest ...;

-- Query mirror status
SELECT * FROM peerdb_internal.mirror_status;
```

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

### S3-Compatible Services Configuration Examples

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Examples of configuring S3-compatible services like MinIO, DigitalOcean Spaces, Backblaze B2, and OVH S3.

```text
- **MinIO**: `endpoint: "https://minio.example.com"`
- **DigitalOcean Spaces**: `region: "nyc3"` + standard config
- **Backblaze B2**: Use specific endpoint
- **OVH S3**: Custom endpoint URL
```

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

### Start PeerDB Server and Create PostgreSQL Peer

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/README.md

Commands to start the PeerDB server and then create a PostgreSQL peer using the REST gateway.

```bash
# Start server
peerdb-flow \
  --temporal-host-port=localhost:7233 \
  --temporal-namespace=default

# Create PostgreSQL peer
curl -X POST http://localhost:8080/v1/peers/create \
  -d '{
    "peer": {
      "name": "pg_source",
      "type": 3,
      "postgres_config": {
        "host": "localhost",
        "port": 5432,
        "user": "postgres",
        "password": "postgres",
        "database": "source"
      }
    }
  }'
```

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

### Clone and Run PeerDB

Source: https://github.com/peerdb-io/peerdb/blob/main/README.md

Clone the PeerDB repository and run the provided scripts to set up and start the PeerDB services using Docker. Ensure Docker and Docker Compose are installed.

```bash
git clone git@github.com:PeerDB-io/peerdb.git
cd peerdb

# Run docker containers: postgres as catalog, temporal, PeerDB server, PeerDB flow API + workers, PeerDB UI
# Requires docker and docker-compose installed: https://docs.docker.com/engine/install/ 
bash ./run-peerdb.sh
# OR for local development, images will be built locally.
# Requires docker, docker-compose as well as the buf compiler for protobuf generation
# https://buf.build/docs/installation
bash ./generate-protos.sh
bash ./dev-peerdb.sh

# connect to peerdb and query away (Use psql version >=14.0)
psql "port=9900 host=localhost password=peerdb"
```

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

### Install Project Dependencies

Source: https://github.com/peerdb-io/peerdb/blob/main/ui/README.md

Installs all necessary project dependencies using npm ci. Ensure Node.js LTS is installed.

```bash
npm ci
```

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

### Snowflake Peer Configuration Example

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Example JSON configuration for connecting to a Snowflake data warehouse. Includes account details, credentials, and database specifics.

```json
{
  "name": "snowflake_dw",
  "type": "SNOWFLAKE",
  "snowflake_config": {
    "account_id": "ab12345.us-east-1",
    "username": "PEERDB_USER",
    "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
    "database": "ANALYTICS",
    "warehouse": "COMPUTE_WH",
    "role": "ACCOUNTADMIN",
    "query_timeout": 600,
    "s3_integration": "s3_int"
  }
}
```

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

### Start MySQL Binlog Synchronization (File-Position Mode)

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/deep-dive-design-document.md

Initiates binlog synchronization for a MySQL source using a specific binlog file and position. Requires the binlog file name and position to start.

```go
CDC->>Syncer: StartSync(file, position)
```

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

### InitialLoadSummary

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Gets detailed progress of initial snapshot load per table.

```APIDOC
## InitialLoadSummary

### Description
Gets detailed progress of initial snapshot load per table.

### Method
GET

### Endpoint
`/v1/mirrors/cdc/initial_load/{parent_mirror_name}`

### Parameters
#### Path Parameters
- **parent_mirror_name** (string) - yes - Mirror name

### Request Example
```json
{
  "parent_mirror_name": "mirror_name"
}
```

### Response
#### Success Response (200)
- **tableSummaries** (CloneTableSummary[]) - Per-table progress

**CloneTableSummary**:
- **table_name** (string) - Table name
- **start_time** (timestamp) - Snapshot start time
- **num_partitions_completed** (int32) - Partitions finished
- **num_partitions_total** (int32) - Total partitions
- **num_rows_synced** (int64) - Rows synced
- **avg_time_per_partition_ms** (int64) - Average partition time
- **flow_job_name** (string) - Flow name
- **source_table** (string) - Source table name
- **fetch_completed** (bool) - Data fetch complete
- **consolidate_completed** (bool) - Consolidation complete
- **mirror_name** (string) - Parent mirror name

#### Response Example
```json
{
  "tableSummaries": [
    {
      "table_name": "users",
      "start_time": "2023-10-27T09:00:00Z",
      "num_partitions_completed": 10,
      "num_partitions_total": 10,
      "num_rows_synced": 500000,
      "avg_time_per_partition_ms": 100,
      "flow_job_name": "mirror_name",
      "source_table": "source_users",
      "fetch_completed": true,
      "consolidate_completed": true,
      "mirror_name": "mirror_name"
    }
  ]
}
```
```

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

### gRPC Error Response Example

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Illustrates a typical gRPC error response, specifically for a SOURCE_TABLE_MISSING error. This example shows the structure of the status and details fields.

```protobuf
status: FAILED_PRECONDITION
details:
  - ErrorInfo {
      reason: "SOURCE_TABLE_MISSING"
      domain: "peerdb"
      metadata: { ... }
    }
  - PreconditionFailure {
      violations: [
        { type: "SOURCE_TABLE_MISSING", subject: "schema.table" }
      ]
    }
```

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

### Start Tilt Environment

Source: https://github.com/peerdb-io/peerdb/blob/main/README.md

Initiate the Tilt environment to manage local development services and run end-to-end tests. Access the Tilt UI for service status and logs.

```bash
./tilt.sh
```

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

### Start MySQL Binlog Synchronization (GTID Mode)

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/deep-dive-design-document.md

Initiates binlog synchronization for a MySQL source using GTID (Global Transaction Identifiers). Requires a valid GTID set to start.

```go
CDC->>Syncer: StartSyncGTID(gset)
```

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

### QRep Example: Incremental Load with Watermark

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/types.md

Demonstrates a QRep flow for incremental loading from PostgreSQL to BigQuery using a timestamp watermark column.

```yaml
flow_job_name: "pg_to_bq_incremental"
source_name: "pg_prod"
destination_name: "bigquery_analytics"
destination_table_identifier: "analytics_dataset.users"
query: "SELECT * FROM public.users"
watermark_table: "public.users"
watermark_column: "updated_at"
write_mode: {
  write_type: QREP_WRITE_MODE_UPSERT
  upsert_key_columns: ["id"]
}
max_parallel_workers: 8
num_rows_per_partition: 50000
staging_path: "gs://staging/qrep"
```

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

### QRep Flow Partitioning Strategies

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

Examples of SQL queries used for partitioning data in QRep flows based on timestamp, integer ID, or PostgreSQL CTID.

```sql
SELECT * FROM table WHERE created_at BETWEEN ? AND ?
```

```sql
SELECT * FROM table WHERE id BETWEEN ? AND ?
```

```sql
SELECT * FROM table WHERE ctid >= ? AND ctid < ?
```

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

### PeerDB QRep Workflow Diagram

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/peerdb-architecture.md

Illustrates the Query Replication (QRep) workflow, detailing the steps from initial setup and partition division to parallel processing of records and final verification.

```mermaid
flowchart TD
    START([QRepFlowWorkflow]) --> META[SetupMetadataTables]
    META --> PARTITIONS[GetQRepPartitions<br/><i>divide by watermark column ranges</i>]

    PARTITIONS --> PARALLEL[QRepPartitionWorkflow<br/><i>Parallel partition processing</i>]

    subgraph PARALLEL_WORK["Parallel Partition Execution"]
        PULL1[PullQRepRecords<br/><i>partition 1</i>]
        PULL2[PullQRepRecords<br/><i>partition 2</i>]
        PULLN[PullQRepRecords<br/><i>partition N</i>]
        SYNC1[SyncQRepRecords<br/><i>partition 1</i>]
        SYNC2[SyncQRepRecords<br/><i>partition 2</i>]
        SYNCN[SyncQRepRecords<br/><i>partition N</i>]
        PULL1 --> SYNC1
        PULL2 --> SYNC2
        PULLN --> SYNCN
    end

    PARALLEL --> PARALLEL_WORK
    PARALLEL_WORK --> VERIFY[VerifyCompletion]
    VERIFY --> DONE([Complete])
```

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

### Workflow Configuration Example

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

This JSON object defines the configuration for a PostgreSQL to BigQuery incremental replication workflow, specifying source, destination, watermark, and performance tuning parameters.

```json
{
  "flow_job_name": "postgres_to_bq_incremental",
  "source_name": "postgres_prod",
  "destination_name": "bigquery_analytics",
  "destination_table_identifier": "dataset.users",
  
  "watermark_table": "public.users",
  "watermark_column": "updated_at",
  
  "initial_copy_only": false,
  "write_mode": {
    "write_type": 1,
    "upsert_key_columns": ["id"]
  },
  
  "max_parallel_workers": 8,
  "num_rows_per_partition": 100000,
  "wait_between_batches_seconds": 60,
  
  "staging_path": "gs://staging/qrep"
}
```

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

### Example Mirror Error Response

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

This is an example of a gRPC status response indicating a general mirror operation failure. It includes the specific PeerDB reason 'MIRROR_ERROR' and relevant metadata.

```protobuf
status: INVALID_ARGUMENT
details:
  - ErrorInfo {
      reason: "MIRROR_ERROR"
      domain: "peerdb"
      metadata: {
        "mirror_name": "postgres_to_snowflake"
        "error_details": "flow not found"
      }
    }
```

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

### CDC Flow Lifecycle Management

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

Demonstrates the typical lifecycle of a CDC flow, including starting, monitoring, pausing, adding tables, and stopping the flow.

```go
client.CreateCDCFlow(ctx, &CreateCDCFlowRequest{
  ConnectionConfigs: cdcConfig,
  AttachToExisting: false,
})
```

```go
status := client.MirrorStatus(ctx, &MirrorStatusRequest{
  FlowJobName: "my_flow",
  IncludeFlowInfo: true,
})
```

```go
client.FlowStateChange(ctx, &FlowStateChangeRequest{
  FlowJobName: "my_flow",
  RequestedFlowState: STATUS_PAUSED,
})
```

```go
client.FlowStateChange(ctx, &FlowStateChangeRequest{
  FlowJobName: "my_flow",
  RequestedFlowState: STATUS_RUNNING,
  FlowConfigUpdate: &FlowConfigUpdate{
    CdcFlowConfigUpdate: &CDCFlowConfigUpdate{
      AdditionalTables: newTableMappings,
    },
  },
})
```

```go
client.FlowStateChange(ctx, &FlowStateChangeRequest{
  FlowJobName: "my_flow",
  RequestedFlowState: STATUS_TERMINATED,
  DropMirrorStats: false,
  SkipDestinationDrop: false,
})
```

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

### Handling Under Maintenance Error in Go

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

Provides a Go code example for checking if an error is due to maintenance and implementing retry logic with backoff.

```go
if errors.Is(err, ErrUnderMaintenance) {
  // Retry after backoff
  // Show user "service unavailable" message
}
```

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

### Kubernetes Deployment Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/ARCHITECTURE.md

Example Kubernetes resource definitions for deploying PeerDB components, including StatefulSets for stateful services like Temporal and PostgreSQL, and Deployments for stateless services like the Flow Service and Nexus.

```yaml
StatefulSet: Temporal
StatefulSet: PostgreSQL (Catalog)
Deployment: Flow Service (replicas: 3)
Deployment: Activity Workers (replicas: 5-20)
Deployment: Nexus (replicas: 2)
Service: LoadBalancer for Flow gRPC
Service: LoadBalancer for Nexus SQL
```

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

### PostgreSQL to Snowflake Type Mapping

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example type mapping from PostgreSQL data types to Snowflake data types.

```text
PostgreSQL Type  -> Snowflake Type
bigint           -> BIGINT
text             -> VARCHAR
json             -> VARIANT
boolean          -> BOOLEAN
timestamp        -> TIMESTAMP_NTZ
uuid             -> VARCHAR
```

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

### MySQL Source Connector Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example JSON configuration for connecting to a MySQL database as a source, supporting CDC via binlog. Requires BINLOG_FORMAT=ROW and MySQL 5.7+ or MariaDB 10.1+.

```json
{
  "name": "mysql_prod",
  "type": "MYSQL",
  "mysql_config": {
    "host": "mysql.internal",
    "port": 3306,
    "user": "peerdb_user",
    "password": "mysql_pass",
    "database": "main",
    "flavor": 1,
    "replication_mechanism": 1,
    "require_tls": true,
    "setup": ["SET SESSION max_connections = 10000"]
  }
}
```

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

### MySQL to BigQuery Type Mapping

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example type mapping from MySQL data types to BigQuery data types.

```text
MySQL Type       -> BigQuery Type
BIGINT           -> INT64
VARCHAR(n)       -> STRING
JSON             -> JSON
BOOLEAN          -> BOOL
DATETIME         -> TIMESTAMP
UUID (CHAR 36)   -> STRING
```

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

### CDC Example: PostgreSQL to Snowflake

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/types.md

Illustrates a common usage scenario for FlowConnectionConfigs, detailing parameters for a CDC flow from PostgreSQL to Snowflake.

```yaml
flow_job_name: "postgres_to_snowflake"
source_name: "pg_prod"
destination_name: "snowflake_analytics"
table_mappings: [
  {
    source_table_identifier: "public.users"
    destination_table_identifier: "analytics.users"
    columns: [...]
  }
]
max_batch_size: 10000
idle_timeout_seconds: 60
do_initial_snapshot: true
snapshot_staging_path: "s3://staging/snapshots"
snapshot_max_parallel_workers: 4
cdc_staging_path: "s3://staging/cdc"
publication_name: "peerdb_pub"
replication_slot_name: "peerdb_slot"
```

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

### Create CDC Mirror Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/README.md

Example using curl to create a CDC mirror, specifying connection configurations like source, destination, and table mappings.

```bash
# Create CDC mirror
curl -X POST http://localhost:8080/v1/flows/cdc/create \
  -d '{
    "connection_configs": {
      "flow_job_name": "pg_to_sf",
      "source_name": "pg_source",
      "destination_name": "sf_dest",
      "table_mappings": [{"source_table_identifier": "public.users", ...}],
      "max_batch_size": 10000,
      "do_initial_snapshot": true
    }
  }'
```

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

### PostgreSQL Source Connector Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example JSON configuration for connecting to a PostgreSQL database as a source, suitable for CDC. Ensure the user has replication permissions and PostgreSQL 10+ is used for logical replication.

```json
{
  "name": "prod_postgres",
  "type": "POSTGRES",
  "postgres_config": {
    "host": "db.prod.internal",
    "port": 5432,
    "user": "replication_user",
    "password": "repl_password",
    "database": "main",
    "metadata_schema": "_peerdb_internal",
    "require_tls": true
  }
}
```

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

### Start Maintenance Mode

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

Initiates maintenance mode to pause running workflows. Use this before cluster updates.

```bash
POST /v1/instance/maintenance
{
  "status": "START",
  "use_peerflow_task_queue": false
}
```

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

### MongoDB Connection URIs

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Examples of connection URIs for MongoDB, including standard, SRV, and replica set configurations.

```text
mongodb://user:password@host:27017/database

mongodb+srv://user:password@cluster.mongodb.net/database

mongodb://host1,host2,host3/database
```

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

### Complete State Machine

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/deep-dive-design-document.md

Visual representation of the PeerDB state machine, illustrating the flow of operations from setup to CDC loop and various signal-driven transitions.

```mermaid
stateDiagram-v2
    [*] --> SetupFlow : CDCFlowWorkflow starts

    state SetupFlow {
        [*] --> EnsurePullability
        EnsurePullability --> SetupReplication
        SetupReplication --> SetupTableSchema
        SetupTableSchema --> SetupNormalizedTables
        SetupNormalizedTables --> [*]
    }

    SetupFlow --> SnapshotFlow : if do_initial_snapshot
    SetupFlow --> CDCLoop : if skip snapshot

    state SnapshotFlow {
        [*] --> ExportSnapshot
        ExportSnapshot --> PartitionTables
        PartitionTables --> ParallelPull
        ParallelPull --> Consolidate
        Consolidate --> [*]
    }

    SnapshotFlow --> CDCLoop

    state CDCLoop {
        [*] --> Pull
        Pull --> Sync
        Sync --> Normalize
        Normalize --> Pull : next batch
        Normalize --> [*] : completed or cancelled
    }
    note right of CDCLoop : Signals (pause, terminate, resync,
config updates) handled in parallel
via Temporal selector

    CDCLoop --> Paused : PauseSignal
    Paused --> CDCLoop : ResumeSignal

    CDCLoop --> TableAddition : add_tables signal
    state TableAddition {
        [*] --> ValidateNoOverlap
        ValidateNoOverlap --> AlterPublication
        AlterPublication --> ChildCDCWorkflow
        ChildCDCWorkflow --> MergeTableMappings
        MergeTableMappings --> [*]
    }
    TableAddition --> CDCLoop

    CDCLoop --> TableRemoval : remove_tables signal
    state TableRemoval {
        [*] --> AlterPubRemove
        AlterPubRemove --> RemoveFromRaw
        RemoveFromRaw --> RemoveFromCatalog
        RemoveFromCatalog --> [*]
    }
    TableRemoval --> CDCLoop

    CDCLoop --> Resync : ResyncSignal
    state Resync {
        [*] --> RenameTablesResync
        RenameTablesResync --> RerunSnapshot
        RerunSnapshot --> RenameBack
        RenameBack --> [*]
    }
    Resync --> CDCLoop

    CDCLoop --> Terminated : TerminateSignal
    Terminated --> DropFlow
    DropFlow --> [*]

    CDCLoop --> Failed : unrecoverable error
    Failed --> [*]

    CDCLoop --> Completed : InitialSnapshotOnly
    Completed --> [*]

```

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

### S3 Configuration for ClickHouse

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Example JSON snippet for configuring S3 staging details when using ClickHouse. Specifies the S3 bucket, path, region, and codec for data staging.

```json
{
  "s3": {
    "url": "s3://my-bucket/staging",
    "region": "us-east-1",
    "codec": "SNAPPY"
  }
}
```

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

### MongoDB Source Connector Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example JSON configuration for connecting to a MongoDB replica set as a source, utilizing change streams for CDC. Requires MongoDB 4.0+ and read permissions on all source collections.

```json
{
  "name": "mongo_prod",
  "type": "MONGO",
  "mongo_config": {
    "uri": "mongodb+srv://user:pass@cluster.mongodb.net/dbname",
    "read_preference": 1,
    "disable_tls": false
  }
}
```

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

### BigQuery Destination Connector Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example JSON configuration for a BigQuery destination connector. Requires a service account with specific GCP roles for data editing and GCS staging.

```json
{
  "name": "bigquery_analytics",
  "type": "BIGQUERY",
  "bigquery_config": {
    "auth_type": "service_account",
    "project_id": "my-gcp-project",
    "private_key_id": "key_id",
    "private_key": "-----BEGIN PRIVATE KEY-----\n...",
    "client_email": "peerdb@my-gcp-project.iam.gserviceaccount.com",
    "client_id": "123456789",
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://oauth2.googleapis.com/token",
    "auth_provider_x509_cert_url": "...",
    "client_x509_cert_url": "...",
    "dataset_id": "analytics"
  }
}
```

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

### Snowflake Storage Integration Setup

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

SQL commands to create a Snowflake storage integration for S3 staging. This integration allows Snowflake to access data in an S3 bucket.

```sql
CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake'
  STORAGE_ALLOWED_LOCATIONS = ('s3://bucket/path');

GRANT CREATE INTEGRATION ON SCHEMA PUBLIC TO ROLE accountadmin;
GRANT USAGE ON INTEGRATION s3_int TO ROLE accountadmin;
```

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

### CDC State Machine Transitions

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/ARCHITECTURE.md

Visualizes the state transitions for the Change Data Capture (CDC) process in PeerDB, from initial setup through running, pausing, resyncing, and termination.

```text
              ┌─────────────┐
              │   SETUP     │ ← Create pub/slot
              │             │
              └─────────────┘
                     ↓
              ┌─────────────┐
    ┌────────→│  SNAPSHOT   │ ← Initial load
    │         │             │
    │         └─────────────┘
    │                ↓
    │         ┌─────────────┐
    │         │   RUNNING   │ ← CDC replication
    │         │             │
    │         └─────────────┘
    │              ↑ ↓
    │              ↓ ↑
    │         ┌─────────────┐
    │         │   PAUSED    │ ← User pause/resume
    │         │             │
    │         └─────────────┘
    │              ↓
    │         ┌─────────────┐
    └────────→│  RESYNC     │ ← Resync operation
              │             │
              └─────────────┘
                     ↓
              ┌─────────────┐
              │ TERMINATING │ ← User terminate
              │             │
              └─────────────┘
                     ↓
              ┌─────────────┐
              │ TERMINATED  │ ← Cleanup done
              │             │
              └─────────────┘
```

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

### Example PeerDB Error Request using curl

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

Demonstrates a curl command to trigger a PeerDB error by attempting to create a flow with a missing source table. This request is designed to elicit a precondition failure.

```bash
curl -X POST http://localhost:8080/v1/flows/cdc/create \
  -d '{
    "connection_configs": {
      "flow_job_name": "test",
      "source_name": "postgres",
      "destination_name": "snowflake",
      "table_mappings": [
        {"source_table_identifier": "public.missing_table", ...}
      ]
    }
  }'
```

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

### Preview Help Output Command

Source: https://github.com/peerdb-io/peerdb/blob/main/flow/switchboard/mongodb/README.md

Use this command to preview the help output without needing a running MongoDB server. It's useful for testing the help text generation.

```bash
go test . -run TestPrintAllHelp -v
```

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

### Structured JSON Log Example

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

An example of a structured JSON log entry generated by PeerDB, including timestamp, log level, message, and flow-specific details like flow name, batch ID, rows processed, and duration.

```json
{
  "time": "2024-01-01T12:00:00Z",
  "level": "INFO",
  "msg": "CDC sync completed",
  "flow_name": "postgres_to_snowflake",
  "batch_id": 123,
  "rows": 10000,
  "duration_ms": 5000
}
```

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

### GetInstanceInfo

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Gets the current status of the PeerDB instance. The request is an empty message.

```APIDOC
## GetInstanceInfo

### Description
Gets instance status.

### Method
GET

### Endpoint
/v1/instance/info

### Response
#### Success Response (200)
- **status** (InstanceStatus) - READY (1) or MAINTENANCE (3)
```

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

### CDCTableTotalCounts

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Gets row count statistics per table for a CDC mirror.

```APIDOC
## CDCTableTotalCounts

### Description
Gets row count statistics per table for a CDC mirror.

### Method
GET

### Endpoint
`/v1/mirrors/cdc/table_total_counts/{flow_job_name}`

### Parameters
#### Path Parameters
- **flow_job_name** (string) - yes - Mirror name

### Request Example
```json
{
  "flow_job_name": "mirror_name"
}
```

### Response
#### Success Response (200)
- **total_data** (CDCRowCounts) - Aggregate counts
- **tables_data** (CDCTableRowCounts[]) - Per-table breakdown

**CDCRowCounts**:
- **total_count** (int64) - Total rows
- **inserts_count** (int64) - Insert operations
- **updates_count** (int64) - Update operations
- **deletes_count** (int64) - Delete operations

**CDCTableRowCounts**:
- **table_name** (string) - Table name
- **counts** (CDCRowCounts) - Row counts for this table

#### Response Example
```json
{
  "total_data": {
    "total_count": 1000000,
    "inserts_count": 900000,
    "updates_count": 90000,
    "deletes_count": 10000
  },
  "tables_data": [
    {
      "table_name": "users",
      "counts": {
        "total_count": 500000,
        "inserts_count": 450000,
        "updates_count": 45000,
        "deletes_count": 5000
      }
    }
  ]
}
```
```

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

### GetCDCBatches

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Retrieves CDC batches with pagination. Supports both GET and POST methods.

```APIDOC
## GetCDCBatches

### Description
Retrieves CDC batches with pagination. Supports both GET and POST methods.

### Method
GET / POST

### Endpoint
GET: `/v1/mirrors/cdc/batches/{flow_job_name}`
POST: `/v1/mirrors/cdc/batches`

### Parameters
#### Path Parameters
- **flow_job_name** (string) - yes - Mirror name

#### Query Parameters
- **limit** (uint32) - no - Number of batches to return
- **ascending** (bool) - no - Order ascending by ID
- **before_id** (int64) - no - Return batches before this ID
- **after_id** (int64) - no - Return batches after this ID

### Request Example
```json
{
  "flow_job_name": "mirror_name",
  "limit": 100,
  "ascending": true,
  "before_id": 12345,
  "after_id": 67890
}
```

### Response
#### Success Response (200)
- **cdc_batches** (CDCBatch[]) - Array of batches
- **total** (int32) - Total batch count
- **page** (int32) - Current page number

#### Response Example
```json
{
  "cdc_batches": [
    {
      "id": 1,
      "timestamp": "2023-10-27T10:00:00Z",
      "operation": "INSERT",
      "data": {}
    }
  ],
  "total": 1000,
  "page": 1
}
```
```

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

### QRep Partitioning Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Adjust QRep partitioning for worker efficiency and source load management.

```json
{
  "num_rows_per_partition": 500000,  // Large partitions = fewer workers needed
  "max_parallel_workers": 8,         // Parallel partitions
  "wait_between_batches_seconds": 5  // Reduce source load spikes
}
```

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

### TotalRowsSyncedByMirror

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Gets total rows synced, with separate CDC and initial load counts.

```APIDOC
## TotalRowsSyncedByMirror

### Description
Gets total rows synced, with separate CDC and initial load counts.

### Method
GET

### Endpoint
`/v1/mirrors/total_rows_synced/{flow_job_name}`

### Parameters
#### Path Parameters
- **flow_job_name** (string) - yes - Mirror name

#### Query Parameters
- **exclude_cdc** (bool) - no - Only count initial load
- **exclude_initial_load** (bool) - no - Only count CDC

### Request Example
```json
{
  "flow_job_name": "mirror_name",
  "exclude_cdc": false,
  "exclude_initial_load": false
}
```

### Response
#### Success Response (200)
- **totalCountCDC** (int64) - Rows from CDC
- **totalCountInitialLoad** (int64) - Rows from initial snapshot
- **totalCount** (int64) - Combined total

#### Response Example
```json
{
  "totalCountCDC": 800000,
  "totalCountInitialLoad": 200000,
  "totalCount": 1000000
}
```
```

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

### Fake PostgreSQL Server Parameters

Source: https://github.com/peerdb-io/peerdb/blob/main/flow/switchboard/README.md

Provides a map of fake PostgreSQL parameters required by psql during startup. These are essential for compatibility, though psql does not validate them.

```go
map[string]string{
			"server_version": "16.0-yourdb-proxy",
			"server_encoding": "UTF8",
			"client_encoding": "UTF8",
			"DateStyle": "ISO, MDY",
			"TimeZone": "UTC",
			"integer_datetimes": "on",
			"standard_conforming_strings": "on",
		}
```

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

### MirrorStatus

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Gets detailed status of a specific mirror. It takes a MirrorStatusRequest and returns a MirrorStatusResponse.

```APIDOC
## POST /v1/mirrors/status

### Description
Gets detailed status of a specific mirror.

### Method
POST

### Endpoint
/v1/mirrors/status

### Parameters
#### Request Body
- **flow_job_name** (`string`) - Required - Mirror/flow name
- **include_flow_info** (`bool`) - Optional - Include full flow configuration
- **exclude_batches** (`bool`) - Optional - Exclude CDC batch details

### Response
#### Success Response (200)
- **flow_job_name** (`string`) - Mirror name
- **status** (oneof) - Either qrep_status or cdc_status
- **current_flow_state** (`peerdb_flow.FlowStatus`) - Current status (RUNNING, PAUSED, etc.)
- **created_at** (`timestamp`) - Mirror creation time

### Response Example (CDC Details)
```json
{
  "flow_job_name": "my_mirror",
  "cdc_status": {
    "config": {},
    "snapshot_status": {},
    "cdc_batches": [
      {
        "start_lsn": 12345,
        "end_lsn": 12346,
        "num_rows": 100,
        "start_time": "2023-10-27T10:00:00Z",
        "end_time": "2023-10-27T10:01:00Z",
        "batch_id": 1
      }
    ],
    "source_type": "POSTGRES",
    "destination_type": "MYSQL",
    "rows_synced": 10000
  },
  "current_flow_state": "RUNNING",
  "created_at": "2023-10-26T09:00:00Z"
}
```
```

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

### Preview Switchboard Help Output

Source: https://github.com/peerdb-io/peerdb/blob/main/flow/switchboard/README.md

Generate help output for Switchboard's MongoDB wire commands without needing a running server. This is useful for previewing available commands and their documentation links.

```go
go test ./mongodb/ -run TestPrintAllHelp -v
```

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

### TimestampPartitionRange Definition

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/types.md

Defines a range for timestamp watermark columns. The range is inclusive of both start and end values.

```protobuf
message TimestampPartitionRange {
  google.protobuf.Timestamp start = 1;
  google.protobuf.Timestamp end = 2;
}
```

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

### Creating a Failed Precondition API Error with Details

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

Demonstrates how to construct a gRPC error with a 'FailedPrecondition' status, including specific details about missing source tables.

```go
import (
  "github.com/PeerDB-io/peerdb/flow/cmd"
  "github.com/PeerDB-io/peerdb/flow/pkg/common"
  "google.golang.org/genproto/googleapis/rpc/errdetails"
)

// Create error with precondition failure details
tables := []common.QualifiedTable{
  {Namespace: "public", Table: "users"},
  {Namespace: "public", Table: "orders"},
}

failure := cmd.NewSourceTableMissingPreconditionFailure(tables)
errorInfo := cmd.NewSourceTableMissingErrorInfo()

err := cmd.NewFailedPreconditionApiError(
  fmt.Errorf("source tables missing"),
  errorInfo,
  failure,
)

// Return as gRPC error
return nil, err
```

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

### Logging Configuration Environment Variables

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Configure PeerDB's structured JSON logging using environment variables.

```markdown
| Variable | Default | Description |
|---|---|---|
| `LOG_LEVEL` | `info` | Minimum log level (debug/info/warn/error) |
| `SLOG_JSON` | — | Output JSON logs (auto-enabled) |
```

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

### Configure PeerDB Flow Service

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Use CLI flags to set Temporal server address, namespace, concurrency limits, and enable profiling or OpenTelemetry metrics and traces.

```bash
peerdb-flow \
  --temporal-host-port=temporal.internal:7233 \
  --temporal-namespace=peerdb \
  --enable-otel-metrics \
  --enable-otel-traces \
  --pprof-port=6060
```

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

### IntPartitionRange Definition

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/types.md

Specifies a range for integer (BIGINT) watermark columns. The range is inclusive of both start and end values.

```protobuf
message IntPartitionRange {
  int64 start = 1;
  int64 end = 2;
}
```

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

### QRep Flow Configuration Options

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/configuration.md

Configuration options for QRep flows, including source, destination, query, and partitioning settings.

```markdown
| Option | Type | Default | Description |
|---|---|---|---|
| flow_job_name | string | — | Unique flow name |
| source_name | string | — | Source peer name |
| destination_name | string | — | Destination peer name |
| destination_table_identifier | string | — | Target table name |
| query | string | — | Custom SQL query (or auto-generated) |
| watermark_table | string | — | Table for watermark column |
| watermark_column | string | — | Column for incremental tracking |
| initial_copy_only | bool | false | Only do initial copy, no incremental |
| max_parallel_workers | uint32 | 4 | Parallel partition processing |
| wait_between_batches_seconds | uint32 | 0 | Delay between batch partitions |
| write_mode | QRepWriteMode | — | Write strategy (APPEND/UPSERT/OVERWRITE) |
| staging_path | string | — | Staging location (gs://, s3://, or local) |
| num_rows_per_partition | uint32 | 100000 | Rows per partition |
| num_partitions_override | uint32 | — | Override partition count |
| setup_watermark_table_on_destination | bool | false | Create watermark table on dest |
| dst_table_full_resync | bool | false | Resync with _peerdb_resync suffix |
| soft_delete_col_name | string | — | Column for soft deletes |
| synced_at_col_name | string | — | Column for sync timestamps |
| system | TypeSystem | Q | Type system |
| script | string | — | Transformation script |
| env | map<string,string> | — | Environment variables |
| exclude | string[] | — | Columns to exclude |
| columns | ColumnSetting[] | — | Per-column transformations |
| flags | string[] | — | Connector capability flags |
| add_null_partition | bool | false | Create partition for NULL watermark values |
```

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

### Get Maintenance Status

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Retrieves the current maintenance status of the instance. This endpoint is useful for monitoring ongoing maintenance operations.

```http
GET /v1/instance/maintenance/status
```

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

### QRepConfig for Query-based Replication

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/types.md

Configuration for Query-based Replication (QRep) flows, supporting full table or watermark-based incremental loads. Defines parameters for extracting and replicating data via queries.

```proto
message QRepConfig {
  string flow_job_name = 1;
  string destination_table_identifier = 4;
  string query = 5;
  string watermark_table = 6;
  string watermark_column = 7;
  
  bool initial_copy_only = 8;
  uint32 max_parallel_workers = 9;
  uint32 wait_between_batches_seconds = 10;
  
  QRepWriteMode write_mode = 11;
  string staging_path = 12;
  
  uint32 num_rows_per_partition = 13;
  uint32 num_partitions_override = 29;
  
  bool setup_watermark_table_on_destination = 14;
  bool dst_table_full_resync = 15;
  
  string synced_at_col_name = 16;
  string soft_delete_col_name = 17;
  
  TypeSystem system = 18;
  string script = 19;
  
  string source_name = 20;
  string destination_name = 21;
  string snapshot_name = 23;
  
  map<string, string> env = 24;
  string parent_mirror_name = 25;
  repeated string exclude = 26;
  repeated ColumnSetting columns = 27;
  uint32 version = 28;
  
  peerdb_peers.DBType source_type = 30;
  repeated string flags = 31;
  bool add_null_partition = 32;
}
```

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

### Run Local End-to-End Tests with MySQL

Source: https://github.com/peerdb-io/peerdb/blob/main/README.md

Execute local end-to-end tests for MySQL to ClickHouse synchronization. Ensure your .env file is configured and PeerDB services are running.

```bash
cd flow
go clean -cache
go test -v -run TestGenericCH_MySQL ./e2e/
```

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

### Flow Status Endpoint Request

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/workflows.md

Example POST request to the /v1/mirrors/status endpoint to retrieve the status and information about a specific replication flow.

```json
{
  "flow_job_name": "my_flow",
  "include_flow_info": true,
  "exclude_batches": false
}
```

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

### Get Mirror Status (REST Gateway)

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/README.md

Retrieves the status of data mirrors using the REST gateway. This corresponds to the gRPC MirrorStatus method.

```APIDOC
## GET /v1/mirrors/status

### Description
Retrieves the current status of all data mirrors.

### Method
GET

### Endpoint
`/v1/mirrors/status`

### Response
#### Success Response (200)
- **status** (object) - Details about mirror statuses.
  - (Fields depend on the actual response structure, not fully detailed in source)
```

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

### Package Layout for MongoDB Compiler

Source: https://github.com/peerdb-io/peerdb/blob/main/flow/switchboard/mongodb/README.md

Provides an overview of the directory structure for the MongoDB wire command compiler package. It lists the main Go files and their responsibilities.

```text
mongodb/
├── allowlist.go       Wire command allowlist, validation, comment support lookup
├── allowlist_test.go
├── mongodb.go         Compile() entry point, ExecSpec, error types, help text
└── mongodb_test.go
```

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

### Example Invalid Peer Response

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

Illustrates a JSON response indicating an invalid peer configuration or connection failure, including a status and a descriptive message.

```json
{
  "status": "INVALID",
  "message": "connect: connection refused (connect to localhost:5432)"
}
```

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

### Maintenance

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/endpoints.md

Starts or ends a maintenance window for the PeerDB instance. Requires a `MaintenanceRequest` specifying the status and an optional flag for using the peerflow task queue.

```APIDOC
## Maintenance

### Description
Starts or ends a maintenance window.

### Method
POST

### Endpoint
/v1/instance/maintenance

### Parameters
#### Request Body
- **status** (MaintenanceStatus) - Required - START (1) or END (2)
- **use_peerflow_task_queue** (bool) - Optional - Use peerflow task queue

### Response
#### Success Response (200)
- **workflow_id** (string) - Temporal workflow ID
- **run_id** (string) - Temporal run ID
```

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

### PeerDB SQL Commands for Management

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/peerdb-architecture.md

Use these SQL commands to manage PeerDB resources like peers and mirrors through the PostgreSQL-compatible interface.

```sql
CREATE PEER my_source FROM POSTGRES WITH (...);
CREATE MIRROR my_mirror FROM my_source TO my_dest WITH (...);
ALTER MIRROR my_mirror ADD TABLES ...;
SELECT * FROM mirror_status;
DROP MIRROR my_mirror;
```

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

### Mirror Operation Error Info Helper

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/errors.md

Shows how to create an `ErrorInfo` object for mirror operations, including metadata like the mirror name and operation type.

```go
// Mirror operation error
func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo
// Example:
info := NewMirrorErrorInfo(map[string]string{
  "mirror_name": "my_flow",
  "operation": "state_change",
})
```

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

### CDC Flow Workflow Diagram

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/peerdb-architecture.md

Visualizes the Change Data Capture (CDC) workflow, illustrating the sequence of operations from setup to the continuous CDC loop.

```mermaid
flowchart TD
    START([CDCFlowWorkflow]) --> SETUP

    subgraph SETUP["1. SetupFlowWorkflow (child)"]
        S1[EnsurePullability<br/><i>verify source permissions</i>]
        S2[SetupReplication<br/><i>create slot + publication PG<br/>binlog position MySQL</i>]
        S3[SetupTableSchema<br/><i>fetch source schemas, store in catalog</i>]
        S4[SetupNormalizedTable<br/><i>create destination tables</i>]
        S1 --> S2 --> S3 --> S4
    end

    SETUP --> SNAPSHOT

    subgraph SNAPSHOT["2. SnapshotFlowWorkflow (child, dedicated task queue)"]
        SN1[ExportTxSnapshot<br/><i>PG: pg_export_snapshot<br/>MySQL: binlog pos<br/>Mongo: timestamp</i>]
        SN2[PartitionTables<br/><i>break into CTID/cursor ranges</i>]
        SN3[PullPartitions<br/><i>parallel reads from source</i>]
        SN4[SyncPartitions<br/><i>AVRO to staging then destination</i>]
        SN5[Consolidate<br/><i>merge partitioned data</i>]
        SN1 --> SN2 --> SN3 --> SN4 --> SN5
    end

    SNAPSHOT --> CDC_LOOP

    subgraph CDC_LOOP["3. CDC Loop (continuous)"]
        direction TB
        C1["SyncFlow Activity<br/><i>Pull → Sync → Normalize</i>"]
        C_SIG["Signal Handling<br/><i>pause, terminate,<br/>resync, config updates</i>"]
        C1 ---|"runs in parallel"| C_SIG
        C_SIG -->|pause| PAUSED([Paused])
        C_SIG -->|terminate| TERM([Terminated])
        C_SIG -->|resync| RESYNC([Resync])
    end

    Note1["Checkpoint advancing happens independently<br/>from Normalize — source continues reading<br/>into S3 staging for ~24h if destination has issues"]

```

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

### Connect to Switchboard with psql

Source: https://github.com/peerdb-io/peerdb/blob/main/flow/switchboard/README.md

Use any PostgreSQL client to connect to Switchboard. The 'database' parameter specifies the target peer. Connect to the 'catalog' peer for special access.

```sh
psql "host=localhost port=5732 dbname=my_peer"
```

```sh
psql "host=localhost port=5732 dbname=catalog"
```

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

### Run MySQL Generic Tests

Source: https://github.com/peerdb-io/peerdb/blob/main/README.md

Execute MySQL generic tests after setting up the environment and ensuring the .env file is configured. This command cleans the Go cache and runs the specified tests.

```bash
go clean -cache; go test -v -run TestGenericCH_MySQL ./e2e/ # Some MySQL generic tests
```

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

### PeerDB CDC Stream State Diagram

Source: https://github.com/peerdb-io/peerdb/blob/main/docs/peerdb-architecture.md

Visualizes the state transitions for a CDC stream within PeerDB, including setup, snapshotting, running, pausing, and termination.

```mermaid
stateDiagram-v2
    [*] --> STATUS_SETUP
    STATUS_SETUP --> STATUS_SNAPSHOT
    STATUS_SNAPSHOT --> STATUS_RUNNING

    STATUS_RUNNING --> STATUS_PAUSED : pause signal
    STATUS_PAUSED --> STATUS_RUNNING : resume signal

    STATUS_RUNNING --> STATUS_TERMINATED : terminate signal
    STATUS_RUNNING --> STATUS_FAILED : unrecoverable error
    STATUS_RUNNING --> STATUS_RESYNC : resync signal<br/>recreate dest tables
    STATUS_RUNNING --> STATUS_MODIFYING : add/remove tables

    STATUS_MODIFYING --> STATUS_RUNNING : modification complete
    STATUS_RESYNC --> STATUS_SNAPSHOT : restart from snapshot
```

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

### MinIO S3-Compatible Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Configuration for using an S3-compatible service like MinIO. Requires endpoint, credentials, and bucket path.

```json
{
  "name": "minio_local",
  "type": "S3",
  "s3_config": {
    "url": "s3://bucket/path",
    "endpoint": "http://minio:9000",
    "access_key_id": "minioadmin",
    "secret_access_key": "minioadmin",
    "codec": 0
  }
}
```

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

### Snowflake Destination Connector Configuration

Source: https://github.com/peerdb-io/peerdb/blob/main/_autodocs/peer-connectors.md

Example JSON configuration for a Snowflake destination connector. Ensure the private key is in PKCS8 format and the S3 integration is correctly set up.

```json
{
  "name": "snowflake_dw",
  "type": "SNOWFLAKE",
  "snowflake_config": {
    "account_id": "ab12345.us-east-1",
    "username": "PEERDB_USER",
    "private_key": "-----BEGIN PRIVATE KEY-----\n...",
    "database": "ANALYTICS",
    "warehouse": "COMPUTE_WH",
    "role": "ANALYST",
    "query_timeout": 600,
    "s3_integration": "peerdb_s3",
    "metadata_schema": "_PEERDB_INTERNAL"
  }
}
```