Repository: xataio/pgstream
Pull request #307 (2025-04-29 12:09:56)

[CLI] Add status command
This PR adds a status command to the CLI, which will help check and validate the pgstream initialisation and the config without the need to perform multiple manual checks on the source database. A `--json` flag allows to visualise the status output as JSON.

```sh
./pgstream status --help
Checks the status of pgstream initialisation and provided configuration

Usage:
  pgstream status [flags]

Examples:

	pgstream status -c pg2pg.env
	pgstream status --postgres-url <postgres-url> --replication-slot <replication-slot-name>
	pgstream status -c pg2pg.yaml --json


Flags:
  -h, --help                      help for status
      --json                      Output the status in JSON format
      --postgres-url string       Source postgres URL where pgstream has been initialised
      --replication-slot string   Name of the postgres replication slot created by pgstream on the source url

Global Flags:
  -c, --config string      .env or .yaml config file to use with pgstream if any
      --log-level string   log level for the application. One of trace, debug, info, warn, error, fatal, panic (default "debug")
```

Example outputs:

- Without errors
```sh
./pgstream status -c snapshot2pg.env
using config file: snapshot2pg.env
 SUCCESS  pgstream status check encountered no issues
Initialisation status:
 - Pgstream schema exists: true
 - Pgstream schema_log table exists: true
 - Migration current version: 7
 - Migration status: success
 - Replication slot name: pgstream_postgres_slot
 - Replication slot plugin: wal2json
 - Replication slot database: postgres
Config status:
 - Valid: true
Transformation rules status:
 - Valid: true
Source status:
 - Reachable: true
```
```sh
./pgstream status -c snapshot2pg.env --json
using config file: snapshot2pg.env
 SUCCESS  pgstream status check encountered no issues
{
	"Init": {
		"PgstreamSchema": {
			"SchemaExists": true,
			"SchemaLogTableExists": true,
			"Errors": null
		},
		"Migration": {
			"Version": 7,
			"Dirty": false,
			"Errors": null
		},
		"ReplicationSlot": {
			"Name": "pgstream_postgres_slot",
			"Plugin": "wal2json",
			"Database": "postgres",
			"Errors": null
		}
	},
	"Config": {
		"Valid": true,
		"Errors": null
	},
	"TransformationRules": {
		"Valid": true,
		"Errors": null
	},
	"Source": {
		"Reachable": true,
		"Errors": null
	}
}
```

- With errors

```sh 
./pgstream status -c snapshot2pg.env
using config file: snapshot2pg.env
 WARNING  pgstream status check identified issues with transformation_rules
Initialisation status:
 - Pgstream schema exists: true
 - Pgstream schema_log table exists: true
 - Migration current version: 7
 - Migration status: success
 - Replication slot name: pgstream_postgres_slot
 - Replication slot plugin: wal2json
 - Replication slot database: postgres
Config status:
 - Valid: true
Transformation rules status:
 - Valid: false
 - Errors: [column id of table "public"."test" has no transformer configured]
Source status:
 - Reachable: true
```

```sh
./pgstream status -c snapshot2pg.env --json
using config file: snapshot2pg.env
 WARNING  pgstream status check identified issues with transformation_rules
{
	"Init": {
		"PgstreamSchema": {
			"SchemaExists": true,
			"SchemaLogTableExists": true,
			"Errors": null
		},
		"Migration": {
			"Version": 7,
			"Dirty": false,
			"Errors": null
		},
		"ReplicationSlot": {
			"Name": "pgstream_postgres_slot",
			"Plugin": "wal2json",
			"Database": "postgres",
			"Errors": null
		}
	},
	"Config": {
		"Valid": true,
		"Errors": null
	},
	"TransformationRules": {
		"Valid": false,
		"Errors": [
			"column id of table \"public\".\"test\" has no transformer configured"
		]
	},
	"Source": {
		"Reachable": true,
		"Errors": null
	}
}
```

github-actions[bot]: ### Merging this branch will **increase** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 21.55% (**+21.55%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 42 (+1) | 0 | 42 (+1) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 22 (+22) | 0 | 22 (+22) |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 85.57% (**+85.57%**) | 97 (+97) | 83 (+83) | 14 (+14) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch will **increase** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 26.00% (**+26.00%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 46 (+2) | 0 | 46 (+2) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 94.59% (**+94.59%**) | 111 (+111) | 105 (+105) | 6 (+6) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch will **increase** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 26.00% (**+26.00%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 46 (+2) | 0 | 46 (+2) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 94.59% (**+94.59%**) | 111 (+111) | 105 (+105) | 6 (+6) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch will **increase** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 26.00% (**+26.00%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 46 (+2) | 0 | 46 (+2) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 94.59% (**+94.59%**) | 111 (+111) | 105 (+105) | 6 (+6) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/internal/postgres | 1.00% (**-0.01%**) | :thumbsdown: |
| github.com/xataio/pgstream/internal/postgres/mocks | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 39.24% (**+39.24%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 48 (+4) | 0 | 48 (+4) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/internal/postgres/mocks/mock_pg_querier.go | 0.00% (ø) | 13 (+3) | 0 | 13 (+3) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn.go | 0.00% (ø) | 23 (+1) | 0 | 23 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn_pool.go | 0.00% (ø) | 24 (+1) | 0 | 24 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_querier.go | 0.00% (ø) | 2 | 0 | 2 |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 85.17% (**+85.17%**) | 209 (+209) | 178 (+178) | 31 (+31) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/internal/postgres | 1.00% (**-0.01%**) | :thumbsdown: |
| github.com/xataio/pgstream/internal/postgres/mocks | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 38.94% (**+38.94%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 48 (+4) | 0 | 48 (+4) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/internal/postgres/mocks/mock_pg_querier.go | 0.00% (ø) | 13 (+3) | 0 | 13 (+3) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn.go | 0.00% (ø) | 23 (+1) | 0 | 23 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn_pool.go | 0.00% (ø) | 24 (+1) | 0 | 24 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_querier.go | 0.00% (ø) | 2 | 0 | 2 |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 83.57% (**+83.57%**) | 213 (+213) | 178 (+178) | 35 (+35) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/internal/postgres | 1.00% (**-0.01%**) | :thumbsdown: |
| github.com/xataio/pgstream/internal/postgres/mocks | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 38.94% (**+38.94%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 48 (+4) | 0 | 48 (+4) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/internal/postgres/mocks/mock_pg_querier.go | 0.00% (ø) | 13 (+3) | 0 | 13 (+3) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn.go | 0.00% (ø) | 23 (+1) | 0 | 23 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn_pool.go | 0.00% (ø) | 24 (+1) | 0 | 24 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_querier.go | 0.00% (ø) | 2 | 0 | 2 |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 77.65% (**+77.65%**) | 85 (+85) | 66 (+66) | 19 (+19) | :star2: |
| github.com/xataio/pgstream/pkg/stream/stream_status_checker.go | 87.50% (**+87.50%**) | 128 (+128) | 112 (+112) | 16 (+16) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_checker_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/cmd | 0.00% (ø) |  |
| github.com/xataio/pgstream/internal/postgres | 1.00% (**-0.01%**) | :thumbsdown: |
| github.com/xataio/pgstream/internal/postgres/mocks | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/stream | 38.94% (**+38.94%**) | :star2: |
| github.com/xataio/pgstream/tools | 0.00% (ø) |  |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/cmd/root_cmd.go | 0.00% (ø) | 48 (+4) | 0 | 48 (+4) |  |
| github.com/xataio/pgstream/cmd/status_cmd.go | 0.00% (ø) | 35 (+35) | 0 | 35 (+35) |  |
| github.com/xataio/pgstream/internal/postgres/mocks/mock_pg_querier.go | 0.00% (ø) | 13 (+3) | 0 | 13 (+3) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn.go | 0.00% (ø) | 23 (+1) | 0 | 23 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_conn_pool.go | 0.00% (ø) | 24 (+1) | 0 | 24 (+1) |  |
| github.com/xataio/pgstream/internal/postgres/pg_querier.go | 0.00% (ø) | 2 | 0 | 2 |  |
| github.com/xataio/pgstream/pkg/stream/stream_status.go | 77.65% (**+77.65%**) | 85 (+85) | 66 (+66) | 19 (+19) | :star2: |
| github.com/xataio/pgstream/pkg/stream/stream_status_checker.go | 87.50% (**+87.50%**) | 128 (+128) | 112 (+112) | 16 (+16) | :star2: |
| github.com/xataio/pgstream/tools/build-cli-definition.go | 0.00% (ø) | 46 (+3) | 0 | 46 (+3) |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/stream/helper_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_checker_test.go
- github.com/xataio/pgstream/pkg/stream/stream_status_test.go

</details>


Patch:
diff --git a/cli-definition.json b/cli-definition.json
index e035f39..5facd77 100644
--- a/cli-definition.json
+++ b/cli-definition.json
@@ -101,6 +101,31 @@
       "subcommands": [],
       "args": []
     },
+    {
+      "name": "status",
+      "short": "Checks the status of pgstream initialisation and provided configuration",
+      "use": "status",
+      "example": "\n\tpgstream status -c pg2pg.env\n\tpgstream status --postgres-url <postgres-url> --replication-slot <replication-slot-name>\n\tpgstream status -c pg2pg.yaml --json\n\t",
+      "flags": [
+        {
+          "name": "json",
+          "description": "Output the status in JSON format",
+          "default": "false"
+        },
+        {
+          "name": "postgres-url",
+          "description": "Source postgres URL where pgstream has been initialised",
+          "default": ""
+        },
+        {
+          "name": "replication-slot",
+          "description": "Name of the postgres replication slot created by pgstream on the source url",
+          "default": ""
+        }
+      ],
+      "subcommands": [],
+      "args": []
+    },
     {
       "name": "tear-down",
       "short": "It tears down any pgstream setup, removing the replication slot and all the relevant tables/functions/triggers, along with the internal pgstream schema",
diff --git a/cmd/root_cmd.go b/cmd/root_cmd.go
index 33c2d77..7577877 100644
--- a/cmd/root_cmd.go
+++ b/cmd/root_cmd.go
@@ -63,6 +63,11 @@ func Prepare() *cobra.Command {
 	runCmd.Flags().StringSlice("snapshot-tables", nil, "List of tables to snapshot if initial snapshot is required, in the format <schema>.<table>. If not specified, the schema `public` will be assumed. Wildcards are supported")
 	runCmd.Flags().Bool("reset", false, "Wether to reset the target before snapshotting (only for postgres target)")
 
+	// status cmd
+	statusCmd.Flags().String("postgres-url", "", "Source postgres URL where pgstream has been initialised")
+	statusCmd.Flags().String("replication-slot", "", "Name of the postgres replication slot created by pgstream on the source url")
+	statusCmd.Flags().Bool("json", false, "Output the status in JSON format")
+
 	// Flag binding for root cmd
 	rootFlagBinding(rootCmd)
 
@@ -71,7 +76,7 @@ func Prepare() *cobra.Command {
 	rootCmd.AddCommand(tearDownCmd)
 	rootCmd.AddCommand(runCmd)
 	rootCmd.AddCommand(snapshotCmd)
-
+	rootCmd.AddCommand(statusCmd)
 	return rootCmd
 }
 
diff --git a/cmd/status_cmd.go b/cmd/status_cmd.go
new file mode 100644
index 0000000..52fef61
--- /dev/null
+++ b/cmd/status_cmd.go
@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/xataio/pgstream/cmd/config"
+	"github.com/xataio/pgstream/pkg/stream"
+)
+
+var statusCmd = &cobra.Command{
+	Use:    "status",
+	Short:  "Checks the status of pgstream initialisation and provided configuration",
+	PreRun: statusFlagBinding,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		sp, _ := pterm.DefaultSpinner.WithText("checking pgstream status...").Start()
+
+		streamConfig, err := config.ParseStreamConfig()
+		if err != nil {
+			return err
+		}
+
+		statusChecker := stream.NewStatusChecker()
+		status, err := statusChecker.Status(context.Background(), streamConfig)
+		if err != nil {
+			sp.Fail(err.Error())
+			return err
+		}
+
+		statusErrs := status.GetErrors()
+		if len(statusErrs) == 0 {
+			sp.Success("pgstream status check encountered no issues")
+		} else {
+			sp.Warning("pgstream status check identified issues with ", strings.Join(statusErrs.Keys(), ", "))
+		}
+
+		err = printStatus(cmd, status)
+		if err != nil {
+			sp.Fail("failed to format pgstream status")
+			return err
+		}
+
+		return nil
+	},
+	Example: `
+	pgstream status -c pg2pg.env
+	pgstream status --postgres-url <postgres-url> --replication-slot <replication-slot-name>
+	pgstream status -c pg2pg.yaml --json
+	`,
+}
+
+func printStatus(cmd *cobra.Command, status *stream.Status) error {
+	statusStr := status.PrettyPrint()
+	if cmd.Flags().Lookup("json").Value.String() == "true" {
+		var prettyJSON bytes.Buffer
+		statusJSON, err := json.Marshal(status)
+		if err != nil {
+			return err
+		}
+		if err := json.Indent(&prettyJSON, statusJSON, "", "\t"); err != nil {
+			return err
+		}
+		statusStr = prettyJSON.String()
+	}
+
+	fmt.Println(statusStr) //nolint:forbidigo
+	return nil
+}
+
+func statusFlagBinding(cmd *cobra.Command, _ []string) {
+	// to be able to overwrite configuration with flags when yaml config file is
+	// provided
+	viper.BindPFlag("source.postgres.url", cmd.Flags().Lookup("postgres-url"))
+	viper.BindPFlag("source.postgres.replication.replication_slot", cmd.Flags().Lookup("replication-slot"))
+	viper.Set("source.postgres.mode", "replication")
+
+	// to be able to overwrite configuration with flags when env config file is
+	// provided or when no configuration is provided
+	viper.BindPFlag("PGSTREAM_POSTGRES_LISTENER_URL", cmd.Flags().Lookup("postgres-url"))
+	viper.BindPFlag("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL", cmd.Flags().Lookup("postgres-url"))
+	viper.BindPFlag("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME", cmd.Flags().Lookup("replication-slot"))
+}
diff --git a/internal/postgres/mocks/mock_pg_querier.go b/internal/postgres/mocks/mock_pg_querier.go
index 0dd0cca..f34d0d0 100644
--- a/internal/postgres/mocks/mock_pg_querier.go
+++ b/internal/postgres/mocks/mock_pg_querier.go
@@ -15,6 +15,7 @@ type Querier struct {
 	ExecFn                   func(context.Context, uint, string, ...any) (postgres.CommandTag, error)
 	ExecInTxFn               func(context.Context, func(tx postgres.Tx) error) error
 	ExecInTxWithOptionsFn    func(context.Context, uint, func(tx postgres.Tx) error, postgres.TxOptions) error
+	PingFn                   func(context.Context) error
 	CloseFn                  func(context.Context) error
 	execCalls                uint32
 	execInTxWithOptionsCalls uint32
@@ -42,6 +43,13 @@ func (m *Querier) ExecInTxWithOptions(ctx context.Context, fn func(tx postgres.T
 	return m.ExecInTxWithOptionsFn(ctx, uint(atomic.LoadUint32(&m.execInTxWithOptionsCalls)), fn, opts)
 }
 
+func (m *Querier) Ping(ctx context.Context) error {
+	if m.PingFn != nil {
+		return m.PingFn(ctx)
+	}
+	return nil
+}
+
 func (m *Querier) Close(ctx context.Context) error {
 	if m.CloseFn != nil {
 		return m.CloseFn(ctx)
diff --git a/internal/postgres/pg_conn.go b/internal/postgres/pg_conn.go
index 076ca41..9fb90e9 100644
--- a/internal/postgres/pg_conn.go
+++ b/internal/postgres/pg_conn.go
@@ -60,6 +60,10 @@ func (c *Conn) ExecInTxWithOptions(ctx context.Context, fn func(Tx) error, opts
 	return tx.Commit(ctx)
 }
 
+func (c *Conn) Ping(ctx context.Context) error {
+	return mapError(c.conn.Ping(ctx))
+}
+
 func (c *Conn) Close(ctx context.Context) error {
 	return mapError(c.conn.Close(ctx))
 }
diff --git a/internal/postgres/pg_conn_pool.go b/internal/postgres/pg_conn_pool.go
index 4c0c0af..6ced242 100644
--- a/internal/postgres/pg_conn_pool.go
+++ b/internal/postgres/pg_conn_pool.go
@@ -60,6 +60,10 @@ func (c *Pool) ExecInTxWithOptions(ctx context.Context, fn func(Tx) error, opts
 	return tx.Commit(ctx)
 }
 
+func (c *Pool) Ping(ctx context.Context) error {
+	return mapError(c.Pool.Ping(ctx))
+}
+
 func (c *Pool) Close(_ context.Context) error {
 	c.Pool.Close()
 	return nil
diff --git a/internal/postgres/pg_querier.go b/internal/postgres/pg_querier.go
index 6c84d52..8b4f205 100644
--- a/internal/postgres/pg_querier.go
+++ b/internal/postgres/pg_querier.go
@@ -15,6 +15,7 @@ type Querier interface {
 	Exec(ctx context.Context, query string, args ...any) (CommandTag, error)
 	ExecInTx(ctx context.Context, fn func(tx Tx) error) error
 	ExecInTxWithOptions(ctx context.Context, fn func(tx Tx) error, txOpts TxOptions) error
+	Ping(ctx context.Context) error
 	Close(ctx context.Context) error
 }
 
diff --git a/pkg/stream/stream_status.go b/pkg/stream/stream_status.go
new file mode 100644
index 0000000..e84ce45
--- /dev/null
+++ b/pkg/stream/stream_status.go
@@ -0,0 +1,220 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package stream
+
+import (
+	"fmt"
+	"strings"
+)
+
+type Status struct {
+	Init                *InitStatus
+	Config              *ConfigStatus
+	TransformationRules *TransformationRulesStatus
+	Source              *SourceStatus
+}
+
+type ConfigStatus struct {
+	Valid  bool
+	Errors []string
+}
+
+type TransformationRulesStatus struct {
+	Valid  bool
+	Errors []string
+}
+
+type SourceStatus struct {
+	Reachable bool
+	Errors    []string
+}
+
+type InitStatus struct {
+	PgstreamSchema  *SchemaStatus
+	Migration       *MigrationStatus
+	ReplicationSlot *ReplicationSlotStatus
+}
+
+type MigrationStatus struct {
+	Version uint
+	Dirty   bool
+	Errors  []string
+}
+
+type ReplicationSlotStatus struct {
+	Name     string
+	Plugin   string
+	Database string
+	Errors   []string
+}
+
+type SchemaStatus struct {
+	SchemaExists         bool
+	SchemaLogTableExists bool
+	Errors               []string
+}
+
+type StatusErrors map[string][]string
+
+func (se StatusErrors) Keys() []string {
+	keys := make([]string, 0, len(se))
+	for k := range se {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+func (s *Status) GetErrors() StatusErrors {
+	if s == nil {
+		return nil
+	}
+
+	errors := map[string][]string{}
+	if initErrs := s.Init.GetErrors(); len(initErrs) > 0 {
+		errors["init"] = initErrs
+	}
+
+	if s.Source != nil && len(s.Source.Errors) > 0 {
+		errors["source"] = s.Source.Errors
+	}
+
+	if s.Config != nil && len(s.Config.Errors) > 0 {
+		errors["config"] = s.Config.Errors
+	}
+
+	if s.TransformationRules != nil && len(s.TransformationRules.Errors) > 0 {
+		errors["transformation_rules"] = s.TransformationRules.Errors
+	}
+
+	return errors
+}
+
+func (s *Status) PrettyPrint() string {
+	if s == nil {
+		return ""
+	}
+
+	var prettyPrint strings.Builder
+	prettyPrint.WriteString(s.Init.PrettyPrint())
+	prettyPrint.WriteByte('\n')
+	prettyPrint.WriteString(s.Config.PrettyPrint())
+	prettyPrint.WriteByte('\n')
+	prettyPrint.WriteString(s.TransformationRules.PrettyPrint())
+	prettyPrint.WriteByte('\n')
+	prettyPrint.WriteString(s.Source.PrettyPrint())
+
+	return prettyPrint.String()
+}
+
+// GetErrors aggregates all errors from the initialisation status.
+func (is *InitStatus) GetErrors() []string {
+	if is == nil {
+		return nil
+	}
+
+	errors := []string{}
+	if is.PgstreamSchema != nil && len(is.PgstreamSchema.Errors) > 0 {
+		errors = append(errors, is.PgstreamSchema.Errors...)
+	}
+
+	if is.Migration != nil && len(is.Migration.Errors) > 0 {
+		errors = append(errors, is.Migration.Errors...)
+	}
+
+	if is.ReplicationSlot != nil && len(is.ReplicationSlot.Errors) > 0 {
+		errors = append(errors, is.ReplicationSlot.Errors...)
+	}
+
+	return errors
+}
+
+func (is *InitStatus) PrettyPrint() string {
+	if is == nil {
+		return ""
+	}
+
+	var prettyPrint strings.Builder
+	prettyPrint.WriteString("Initialisation status:\n")
+	if is.PgstreamSchema != nil {
+		prettyPrint.WriteString(fmt.Sprintf(" - Pgstream schema exists: %t\n", is.PgstreamSchema.SchemaExists))
+		prettyPrint.WriteString(fmt.Sprintf(" - Pgstream schema_log table exists: %t\n", is.PgstreamSchema.SchemaLogTableExists))
+		if len(is.PgstreamSchema.Errors) > 0 {
+			prettyPrint.WriteString(fmt.Sprintf(" - Pgstream schema errors: %s\n", is.PgstreamSchema.Errors))
+		}
+	}
+
+	if is.Migration != nil {
+		prettyPrint.WriteString(fmt.Sprintf(" - Migration current version: %d\n", is.Migration.Version))
+		prettyPrint.WriteString(fmt.Sprintf(" - Migration status: %s\n", migrationStatus(is.Migration.Dirty)))
+		if len(is.Migration.Errors) > 0 {
+			prettyPrint.WriteString(fmt.Sprintf(" - Migration errors: %s\n", is.Migration.Errors))
+		}
+	}
+
+	if is.ReplicationSlot != nil {
+		prettyPrint.WriteString(fmt.Sprintf(" - Replication slot name: %s\n", is.ReplicationSlot.Name))
+		prettyPrint.WriteString(fmt.Sprintf(" - Replication slot plugin: %s\n", is.ReplicationSlot.Plugin))
+		prettyPrint.WriteString(fmt.Sprintf(" - Replication slot database: %s\n", is.ReplicationSlot.Database))
+		if len(is.ReplicationSlot.Errors) > 0 {
+			prettyPrint.WriteString(fmt.Sprintf(" - Replication slot errors: %s\n", is.ReplicationSlot.Errors))
+		}
+	}
+
+	// trim the last newline character
+	return strings.TrimSuffix(prettyPrint.String(), "\n")
+}
+
+func (ss *SourceStatus) PrettyPrint() string {
+	if ss == nil {
+		return ""
+	}
+
+	var prettyPrint strings.Builder
+	prettyPrint.WriteString("Source status:\n")
+	prettyPrint.WriteString(fmt.Sprintf(" - Reachable: %t\n", ss.Reachable))
+	if len(ss.Errors) > 0 {
+		prettyPrint.WriteString(fmt.Sprintf(" - Errors: %s\n", ss.Errors))
+	}
+
+	// trim the last newline character
+	return strings.TrimSuffix(prettyPrint.String(), "\n")
+}
+
+func (cs *ConfigStatus) PrettyPrint() string {
+	if cs == nil {
+		return ""
+	}
+
+	var prettyPrint strings.Builder
+	prettyPrint.WriteString("Config status:\n")
+	prettyPrint.WriteString(fmt.Sprintf(" - Valid: %t\n", cs.Valid))
+	if len(cs.Errors) > 0 {
+		prettyPrint.WriteString(fmt.Sprintf(" - Errors: %s\n", cs.Errors))
+	}
+
+	// trim the last newline character
+	return strings.TrimSuffix(prettyPrint.String(), "\n")
+}
+
+func (trs *TransformationRulesStatus) PrettyPrint() string {
+	if trs == nil {
+		return ""
+	}
+
+	var prettyPrint strings.Builder
+	prettyPrint.WriteString("Transformation rules status:\n")
+	prettyPrint.WriteString(fmt.Sprintf(" - Valid: %t\n", trs.Valid))
+	if len(trs.Errors) > 0 {
+		prettyPrint.WriteString(fmt.Sprintf(" - Errors: %s\n", trs.Errors))
+	}
+
+	// trim the last newline character
+	return strings.TrimSuffix(prettyPrint.String(), "\n")
+}
+
+func migrationStatus(dirty bool) string {
+	if !dirty {
+		return "success"
+	}
+	return "failed"
+}
diff --git a/pkg/stream/stream_status_checker.go b/pkg/stream/stream_status_checker.go
new file mode 100644
index 0000000..72081e8
--- /dev/null
+++ b/pkg/stream/stream_status_checker.go
@@ -0,0 +1,356 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package stream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"syscall"
+
+	pglib "github.com/xataio/pgstream/internal/postgres"
+	pgmigrations "github.com/xataio/pgstream/migrations/postgres"
+	"github.com/xataio/pgstream/pkg/wal/processor/transformer"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// StatusChecker is responsible for validating the status of the pgstream setup
+// in a PostgreSQL database. It performs checks on the source database
+// connection, initialization status (schema, migrations, replication slot), and
+// transformation rules. It provides detailed status information, including
+// errors, to help diagnose issues with the pgstream configuration and setup.
+type StatusChecker struct {
+	connBuilder          pglib.QuerierBuilder
+	configParser         func(pgURL string) (*pgx.ConnConfig, error)
+	migratorBuilder      func(string) (migrator, error)
+	ruleValidatorBuilder func(context.Context, string) (ruleValidator, error)
+}
+
+type ruleValidator func(rules []transformer.TableRules) (map[string]transformer.ColumnTransformers, error)
+
+type migrator interface {
+	Version() (uint, bool, error)
+	Close() (error, error)
+}
+
+const wal2jsonPlugin = "wal2json"
+
+const (
+	noPgstreamSchemaErrMsg         = "pgstream schema does not exist in the configured postgres database"
+	noPgstreamSchemaLogTableErrMsg = "pgstream schema_log table does not exist in the configured postgres database"
+	noMigrationsTableErrMsg        = "pgstream schema_migrations table does not exist in the configured postgres database"
+	sourceNotProvided              = "source not provided"
+	sourcePostgresNotReachable     = "source postgres not reachable"
+)
+
+func NewStatusChecker() *StatusChecker {
+	return &StatusChecker{
+		connBuilder:     pglib.ConnBuilder,
+		configParser:    pgx.ParseConfig,
+		migratorBuilder: func(pgURL string) (migrator, error) { return newPGMigrator(pgURL) },
+		ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+			validator, err := transformer.NewPostgresTransformerParser(ctx, pgURL)
+			if err != nil {
+				return nil, err
+			}
+			return validator.ParseAndValidate, nil
+		},
+	}
+}
+
+// Status retrieves the overall status of the pgstream setup, including the
+// source database connection status, initialization status (schema, migrations,
+// replication slot), and transformation rules validation status. It returns a
+// detailed status report that includes any errors encountered during the
+// checks, helping to diagnose issues with the pgstream configuration and setup.
+func (s *StatusChecker) Status(ctx context.Context, config *Config) (*Status, error) {
+	sourceStatus, err := s.sourceStatus(ctx, config)
+	if err != nil {
+		return nil, fmt.Errorf("checking source status: %w", err)
+	}
+
+	initStatus, err := s.initStatus(ctx, config.SourcePostgresURL(), config.PostgresReplicationSlot())
+	if err != nil {
+		return nil, fmt.Errorf("checking init status: %w", err)
+	}
+
+	transformationRulesStatus, err := s.transformationRulesStatus(ctx, config)
+	if err != nil {
+		return nil, fmt.Errorf("checking transformation rules status: %w", err)
+	}
+
+	return &Status{
+		Init:                initStatus,
+		Config:              s.configStatus(config),
+		Source:              sourceStatus,
+		TransformationRules: transformationRulesStatus,
+	}, nil
+}
+
+// configStatus validates if the configuration provided is valid.
+func (s *StatusChecker) configStatus(config *Config) *ConfigStatus {
+	if err := config.IsValid(); err != nil {
+		return &ConfigStatus{
+			Valid:  false,
+			Errors: []string{err.Error()},
+		}
+	}
+
+	return &ConfigStatus{
+		Valid: true,
+	}
+}
+
+// sourceStatus validates if the source postgres database is reachable.
+func (s *StatusChecker) sourceStatus(ctx context.Context, config *Config) (*SourceStatus, error) {
+	sourcePostgresURL := config.SourcePostgresURL()
+	if sourcePostgresURL == "" {
+		return &SourceStatus{
+			Reachable: false,
+			Errors:    []string{sourceNotProvided},
+		}, nil
+	}
+
+	conn, err := s.connBuilder(ctx, sourcePostgresURL)
+	if err != nil {
+		if errors.Is(err, syscall.ECONNREFUSED) {
+			return &SourceStatus{
+				Reachable: false,
+				Errors:    []string{fmt.Sprintf("%s: %v", sourcePostgresNotReachable, err)},
+			}, nil
+		}
+		return nil, err
+	}
+	defer conn.Close(context.Background())
+
+	sourceStatus := &SourceStatus{
+		Reachable: true,
+	}
+	if err := conn.Ping(ctx); err != nil {
+		sourceStatus.Reachable = false
+		sourceStatus.Errors = append(sourceStatus.Errors, err.Error())
+	}
+
+	return sourceStatus, nil
+}
+
+// initStatus checks the initialisation status of pgstream in the provided
+// postgres database. If the replicationSlotName is empty, it will check against
+// the default (pgstream_<database>_slot).
+func (s *StatusChecker) initStatus(ctx context.Context, pgURL, replicationSlotName string) (*InitStatus, error) {
+	if pgURL == "" {
+		return nil, errMissingPostgresURL
+	}
+
+	initStatus := &InitStatus{}
+
+	var err error
+	initStatus.PgstreamSchema, err = s.validateSchemaStatus(ctx, pgURL)
+	if err != nil {
+		return nil, fmt.Errorf("validating pgstream schema: %w", err)
+	}
+
+	initStatus.Migration, err = s.validateMigrationStatus(pgURL)
+	if err != nil {
+		return nil, fmt.Errorf("validating pgstream migrations: %w", err)
+	}
+
+	initStatus.ReplicationSlot, err = s.validateReplicationSlotStatus(ctx, pgURL, replicationSlotName)
+	if err != nil {
+		return nil, fmt.Errorf("validating replication slot: %w", err)
+	}
+
+	return initStatus, nil
+}
+
+// transformationRulesStatus validates that the transformation rules provided in
+// the configuration are valid, in line with the validation performed during the
+// pgstream run/snapshot commands.
+func (s *StatusChecker) transformationRulesStatus(ctx context.Context, config *Config) (*TransformationRulesStatus, error) {
+	if config.Processor.Transformer == nil {
+		return nil, nil
+	}
+
+	pgURL := config.SourcePostgresURL()
+	if pgURL == "" {
+		return &TransformationRulesStatus{
+			Valid:  false,
+			Errors: []string{fmt.Sprintf("cannot validate transformer rules: %s", sourceNotProvided)},
+		}, nil
+	}
+
+	status := &TransformationRulesStatus{
+		Valid: true,
+	}
+	validator, err := s.ruleValidatorBuilder(ctx, pgURL)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := validator(config.Processor.Transformer.TransformerRules); err != nil {
+		status.Valid = false
+		switch {
+		case errors.Is(err, syscall.ECONNREFUSED):
+			status.Errors = append(status.Errors, fmt.Sprintf("cannot validate transformer rules: %s", sourcePostgresNotReachable))
+		default:
+			status.Errors = append(status.Errors, err.Error())
+		}
+	}
+
+	return status, nil
+}
+
+// validateMigrationStatus checks the migration status of the pgstream schema,
+// ensuring the number of migrations applied corresponds to the expected ones,
+// and they are not dirty (unsuccessfully applied).
+func (s *StatusChecker) validateMigrationStatus(pgURL string) (*MigrationStatus, error) {
+	migrator, err := s.migratorBuilder(pgURL)
+	if err != nil {
+		switch {
+		case strings.Contains(err.Error(), "failed to open database: no schema"):
+			return &MigrationStatus{
+				Errors: []string{noMigrationsTableErrMsg},
+			}, nil
+		case errors.Is(err, syscall.ECONNREFUSED):
+			return &MigrationStatus{
+				Errors: []string{fmt.Sprintf("cannot validate migration status: %s", sourcePostgresNotReachable)},
+			}, nil
+		default:
+			return nil, fmt.Errorf("error creating postgres migrator: %w", err)
+		}
+	}
+	defer migrator.Close()
+
+	version, dirty, err := migrator.Version()
+	if err != nil {
+		return nil, fmt.Errorf("error getting migration version: %w", err)
+	}
+
+	migrationErrs := func() []string {
+		var errs []string
+		migrationAssets := pgmigrations.AssetNames()
+		if len(migrationAssets)/2 != int(version) {
+			errs = append(errs, fmt.Sprintf("migration version (%d) does not match the number of migration files (%d)", version, len(migrationAssets)/2))
+		}
+
+		if dirty {
+			errs = append(errs, fmt.Sprintf("migration version %d is dirty", version))
+		}
+
+		return errs
+	}()
+
+	return &MigrationStatus{
+		Version: version,
+		Dirty:   dirty,
+		Errors:  migrationErrs,
+	}, nil
+}
+
+// validateSchemaStatus checks if the pgstream schema and schema_log table exist
+// in the configured postgres database.
+func (s *StatusChecker) validateSchemaStatus(ctx context.Context, pgURL string) (*SchemaStatus, error) {
+	conn, err := s.connBuilder(ctx, pgURL)
+	if err != nil {
+		if errors.Is(err, syscall.ECONNREFUSED) {
+			return &SchemaStatus{
+				Errors: []string{fmt.Sprintf("cannot validate schema status: %s", sourcePostgresNotReachable)},
+			}, nil
+		}
+		return nil, err
+	}
+	defer conn.Close(ctx)
+
+	var schemaExists bool
+	err = conn.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')").Scan(&schemaExists)
+	if err != nil && !errors.Is(err, pglib.ErrNoRows) {
+		return nil, fmt.Errorf("checking if pgstream schema exists: %w", err)
+	}
+
+	// if the schema doesn't exist there's no need to check for the schema_log
+	// table as it won't exist either
+	if !schemaExists {
+		return &SchemaStatus{
+			SchemaExists:         false,
+			SchemaLogTableExists: false,
+			Errors:               append([]string{}, noPgstreamSchemaErrMsg),
+		}, nil
+	}
+
+	var schemaLogTableExists bool
+	err = conn.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')").Scan(&schemaLogTableExists)
+	if err != nil && !errors.Is(err, pglib.ErrNoRows) {
+		return nil, fmt.Errorf("checking if pgstream schema_log table exists: %w", err)
+	}
+
+	status := &SchemaStatus{
+		SchemaExists:         schemaExists,
+		SchemaLogTableExists: schemaLogTableExists,
+	}
+
+	if !schemaLogTableExists {
+		status.Errors = append(status.Errors, noPgstreamSchemaLogTableErrMsg)
+	}
+
+	return status, nil
+}
+
+// validateReplicationSlotStatus checks if the replication slot exists in the
+// configured postgres database, and it's correctly configured.
+func (s *StatusChecker) validateReplicationSlotStatus(ctx context.Context, pgURL, replicationSlotName string) (*ReplicationSlotStatus, error) {
+	if replicationSlotName == "" {
+		var err error
+		replicationSlotName, err = getReplicationSlotName(pgURL)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Check the replication slot exists in the configured database by parsing
+	// and extracting the database name from the connection string.
+	cfg, err := s.configParser(pgURL)
+	if err != nil {
+		return nil, fmt.Errorf("parsing postgres connection string: %w", err)
+	}
+	if cfg.Database == "" {
+		cfg.Database = "postgres"
+	}
+
+	conn, err := s.connBuilder(ctx, pgURL)
+	if err != nil {
+		if errors.Is(err, syscall.ECONNREFUSED) {
+			return &ReplicationSlotStatus{
+				Errors: []string{fmt.Sprintf("cannot validate replication slot status: %s", sourcePostgresNotReachable)},
+			}, nil
+		}
+		return nil, err
+	}
+	defer conn.Close(ctx)
+
+	var name, plugin, database string
+	err = conn.QueryRow(ctx, "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1", replicationSlotName).
+		Scan(&name, &plugin, &database)
+	if err != nil && !errors.Is(err, pglib.ErrNoRows) {
+		return nil, fmt.Errorf("retrieving replication slot information: %w", err)
+	}
+
+	replicationErrs := func() []string {
+		if database != cfg.Database {
+			return []string{fmt.Sprintf("replication slot %s does not exist in the configured database", replicationSlotName)}
+		}
+
+		if plugin != wal2jsonPlugin {
+			return []string{fmt.Sprintf("replication slot %s is not using the wal2json plugin", replicationSlotName)}
+		}
+
+		return nil
+	}()
+
+	return &ReplicationSlotStatus{
+		Name:     name,
+		Plugin:   plugin,
+		Database: database,
+		Errors:   replicationErrs,
+	}, nil
+}
diff --git a/tools/build-cli-definition.go b/tools/build-cli-definition.go
index 8397941..993e702 100644
--- a/tools/build-cli-definition.go
+++ b/tools/build-cli-definition.go
@@ -44,7 +44,7 @@ func main() {
 	result := Result{
 		Name:     rootCmd.Name(),
 		Commands: extractCommands(rootCmd.Commands()),
-		Flags:    extractFlags(rootCmd.PersistentFlags()),
+		Flags:    extractFlags([]*pflag.FlagSet{rootCmd.PersistentFlags()}),
 	}
 
 	if err := writeJSONToFile("cli-definition.json", result); err != nil {
@@ -73,25 +73,35 @@ func processCommand(cmd *cobra.Command) Command {
 		Use:         cmd.Use,
 		Example:     cmd.Example,
 		Args:        validateArgs(cmd),
-		Flags:       extractFlags(cmd.Flags()),
+		Flags:       extractFlags([]*pflag.FlagSet{cmd.Flags(), cmd.PersistentFlags()}),
 		Subcommands: extractCommands(cmd.Commands()),
 	}
 }
 
-func extractFlags(flagSet *pflag.FlagSet) []Flag {
-	if flagSet == nil {
+func extractFlags(flagSet []*pflag.FlagSet) []Flag {
+	if len(flagSet) == 0 {
 		return []Flag{}
 	}
 
-	flags := make([]Flag, 0, flagSet.NFlag())
-	flagSet.VisitAll(func(flag *pflag.Flag) {
-		flags = append(flags, Flag{
-			Name:        flag.Name,
-			Shorthand:   flag.Shorthand,
-			Description: flag.Usage,
-			Default:     flag.DefValue,
+	flags := []Flag{}
+	seen := make(map[string]bool) // Track flags by name
+	for _, f := range flagSet {
+		if f == nil {
+			continue
+		}
+		f.VisitAll(func(flag *pflag.Flag) {
+			if !seen[flag.Name] { // Only add flag if not already seen
+				seen[flag.Name] = true
+				flags = append(flags, Flag{
+					Name:        flag.Name,
+					Shorthand:   flag.Shorthand,
+					Description: flag.Usage,
+					Default:     flag.DefValue,
+				})
+			}
 		})
-	})
+	}
+
 	return flags
 }
 


Test patch:
diff --git a/pkg/stream/helper_test.go b/pkg/stream/helper_test.go
new file mode 100644
index 0000000..e53a8ca
--- /dev/null
+++ b/pkg/stream/helper_test.go
@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package stream
+
+type mockMigrator struct {
+	versionFn func() (uint, bool, error)
+	closeFn   func() (error, error)
+}
+
+func (m *mockMigrator) Version() (uint, bool, error) {
+	if m.versionFn != nil {
+		return m.versionFn()
+	}
+	return 0, false, nil
+}
+
+func (m *mockMigrator) Close() (error, error) {
+	if m.closeFn != nil {
+		return m.closeFn()
+	}
+	return nil, nil
+}
diff --git a/pkg/stream/stream_status_checker_test.go b/pkg/stream/stream_status_checker_test.go
new file mode 100644
index 0000000..6076e46
--- /dev/null
+++ b/pkg/stream/stream_status_checker_test.go
@@ -0,0 +1,1249 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package stream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"syscall"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/require"
+	pglib "github.com/xataio/pgstream/internal/postgres"
+	pgmocks "github.com/xataio/pgstream/internal/postgres/mocks"
+	pgmigrations "github.com/xataio/pgstream/migrations/postgres"
+	pgprocessor "github.com/xataio/pgstream/pkg/wal/processor/postgres"
+	"github.com/xataio/pgstream/pkg/wal/processor/transformer"
+	replicationpg "github.com/xataio/pgstream/pkg/wal/replication/postgres"
+)
+
+func TestStatusChecker_Status(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+	migrationVersion := uint(len(pgmigrations.AssetNames()) / 2)
+	const testReplicationSlot = "pgstream_db_slot"
+	const testDB = "db"
+
+	validConfig := &Config{
+		Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+			PostgresURL:         "postgres://user:password@localhost:5432/db",
+			ReplicationSlotName: testReplicationSlot,
+		}}},
+		Processor: ProcessorConfig{
+			Postgres: &PostgresProcessorConfig{
+				BatchWriter: pgprocessor.Config{
+					URL: "postgres://user:password@localhost:7654/db",
+				},
+			},
+			Transformer: &transformer.Config{
+				TransformerRules: []transformer.TableRules{},
+			},
+		},
+	}
+
+	validConfigParser := func(pgURL string) (*pgx.ConnConfig, error) {
+		return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+	}
+
+	validMigratorBuilder := func(s string) (migrator, error) {
+		return &mockMigrator{
+			versionFn: func() (uint, bool, error) {
+				return migrationVersion, false, nil
+			},
+		}, nil
+	}
+
+	validConnBuilder := func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+		return &pgmocks.Querier{
+			QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+				switch sql {
+				case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+					return &pgmocks.Row{ScanFn: func(dest ...any) error {
+						require.Len(t, dest, 1)
+						exists, ok := dest[0].(*bool)
+						require.True(t, ok)
+						*exists = true
+						return nil
+					}}
+				case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+					return &pgmocks.Row{ScanFn: func(dest ...any) error {
+						require.Len(t, dest, 1)
+						exists, ok := dest[0].(*bool)
+						require.True(t, ok)
+						*exists = true
+						return nil
+					}}
+				case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+					require.Len(t, args, 1)
+					require.Equal(t, args[0], testReplicationSlot)
+					return &pgmocks.Row{ScanFn: func(dest ...any) error {
+						require.Len(t, dest, 3)
+
+						name, ok := dest[0].(*string)
+						require.True(t, ok)
+						*name = testReplicationSlot
+
+						plugin, ok := dest[1].(*string)
+						require.True(t, ok)
+						*plugin = wal2jsonPlugin
+
+						db, ok := dest[2].(*string)
+						require.True(t, ok)
+						*db = testDB
+
+						return nil
+					}}
+				default:
+					return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+				}
+			},
+		}, nil
+	}
+
+	validRuleValidatorBuilder := func(context.Context, string) (ruleValidator, error) {
+		return func(rules []transformer.TableRules) (map[string]transformer.ColumnTransformers, error) {
+			return nil, nil
+		}, nil
+	}
+
+	tests := []struct {
+		name                 string
+		connBuilder          pglib.QuerierBuilder
+		migratorBuilder      func(string) (migrator, error)
+		config               *Config
+		ruleValidatorBuilder func(context.Context, string) (ruleValidator, error)
+
+		wantStatus *Status
+		wantErr    error
+	}{
+		{
+			name:                 "ok - all components valid",
+			connBuilder:          validConnBuilder,
+			migratorBuilder:      validMigratorBuilder,
+			config:               validConfig,
+			ruleValidatorBuilder: validRuleValidatorBuilder,
+
+			wantStatus: &Status{
+				Init: &InitStatus{
+					PgstreamSchema: &SchemaStatus{
+						SchemaExists:         true,
+						SchemaLogTableExists: true,
+					},
+					Migration: &MigrationStatus{
+						Version: migrationVersion,
+						Dirty:   false,
+					},
+					ReplicationSlot: &ReplicationSlotStatus{
+						Name:     testReplicationSlot,
+						Plugin:   wal2jsonPlugin,
+						Database: testDB,
+					},
+				},
+				Config: &ConfigStatus{
+					Valid: true,
+				},
+				Source: &SourceStatus{
+					Reachable: true,
+				},
+				TransformationRules: &TransformationRulesStatus{
+					Valid: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - checking source status",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return nil, errTest
+			},
+			migratorBuilder:      validMigratorBuilder,
+			config:               validConfig,
+			ruleValidatorBuilder: validRuleValidatorBuilder,
+
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name:        "error - checking init status",
+			connBuilder: validConnBuilder,
+			migratorBuilder: func(s string) (migrator, error) {
+				return nil, errTest
+			},
+			config:               validConfig,
+			ruleValidatorBuilder: validRuleValidatorBuilder,
+
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name:            "error - checking transformation rules status",
+			connBuilder:     validConnBuilder,
+			migratorBuilder: validMigratorBuilder,
+			config:          validConfig,
+			ruleValidatorBuilder: func(ctx context.Context, s string) (ruleValidator, error) {
+				return nil, errTest
+			},
+
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.connBuilder = tc.connBuilder
+			sc.configParser = validConfigParser
+			sc.migratorBuilder = tc.migratorBuilder
+			sc.ruleValidatorBuilder = tc.ruleValidatorBuilder
+
+			status, err := sc.Status(context.Background(), tc.config)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
+
+func TestStatusChecker_configStatus(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		config     *Config
+		wantStatus *ConfigStatus
+	}{
+		{
+			name: "valid config",
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+				Processor: ProcessorConfig{
+					Postgres: &PostgresProcessorConfig{
+						BatchWriter: pgprocessor.Config{
+							URL: "postgres://user:password@localhost:7654/db",
+						},
+					},
+				},
+			},
+			wantStatus: &ConfigStatus{
+				Valid: true,
+			},
+		},
+		{
+			name: "invalid config",
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &ConfigStatus{
+				Valid:  false,
+				Errors: []string{"need at least one processor configured"},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			status := sc.configStatus(tc.config)
+			require.Equal(t, tc.wantStatus, status)
+		})
+	}
+}
+
+func TestStatusChecker_sourceStatus(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name        string
+		connBuilder pglib.QuerierBuilder
+		config      *Config
+		wantStatus  *SourceStatus
+		wantErr     error
+	}{
+		{
+			name: "ok - source reachable",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					PingFn: func(ctx context.Context) error {
+						return nil
+					},
+				}, nil
+			},
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &SourceStatus{
+				Reachable: true,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - source not provided",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return nil, errors.New("unexpected call to connBuilder")
+			},
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "",
+				}}},
+			},
+			wantStatus: &SourceStatus{
+				Reachable: false,
+				Errors:    []string{sourceNotProvided},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - connection refused",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return nil, syscall.ECONNREFUSED
+			},
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &SourceStatus{
+				Reachable: false,
+				Errors:    []string{fmt.Sprintf("%s: %v", sourcePostgresNotReachable, syscall.ECONNREFUSED)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - ping failure",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					PingFn: func(ctx context.Context) error {
+						return errTest
+					},
+				}, nil
+			},
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &SourceStatus{
+				Reachable: false,
+				Errors:    []string{errTest.Error()},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - unexpected connection error",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return nil, errTest
+			},
+			config: &Config{
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.connBuilder = tc.connBuilder
+
+			status, err := sc.sourceStatus(context.Background(), tc.config)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
+
+func TestStatusChecker_initStatus(t *testing.T) {
+	t.Parallel()
+
+	migrationVersion := uint(len(pgmigrations.AssetNames()) / 2)
+	const testReplicationSlot = "pgstream_db_slot"
+	const testDB = "db"
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name                string
+		connBuilder         pglib.QuerierBuilder
+		configParser        func(pgURL string) (*pgx.ConnConfig, error)
+		migratorBuilder     func(string) (migrator, error)
+		pgurl               string
+		replicationSlotName string
+
+		wantstatus *InitStatus
+		wantErr    error
+	}{
+		{
+			name: "ok",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							require.Len(t, args, 1)
+							require.Equal(t, args[0], testReplicationSlot)
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 3)
+
+								name, ok := dest[0].(*string)
+								require.True(t, ok)
+								*name = testReplicationSlot
+
+								plugin, ok := dest[1].(*string)
+								require.True(t, ok)
+								*plugin = wal2jsonPlugin
+
+								db, ok := dest[2].(*string)
+								require.True(t, ok)
+								*db = testDB
+
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			migratorBuilder: func(s string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return migrationVersion, false, nil
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+
+			wantstatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         true,
+					SchemaLogTableExists: true,
+				},
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     testReplicationSlot,
+					Plugin:   wal2jsonPlugin,
+					Database: testDB,
+				},
+				Migration: &MigrationStatus{
+					Version: migrationVersion,
+					Dirty:   false,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name:    "error - missing postgres URL",
+			pgurl:   "",
+			wantErr: errMissingPostgresURL,
+		},
+		{
+			name: "error - validating schema",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								return errTest
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			migratorBuilder: func(s string) (migrator, error) {
+				return nil, errors.New("unexpected call to migrator builder")
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return nil, errors.New("unexpected call to config parser")
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+
+			wantstatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name: "error - validating migrations",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			migratorBuilder: func(s string) (migrator, error) {
+				return nil, errTest
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return nil, errors.New("unexpected call to config parser")
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+
+			wantstatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name: "error - validating replication slot",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								return errTest
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			migratorBuilder: func(s string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return migrationVersion, false, nil
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+
+			wantstatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.configParser = tc.configParser
+			sc.connBuilder = tc.connBuilder
+			sc.migratorBuilder = tc.migratorBuilder
+
+			status, err := sc.initStatus(context.Background(), tc.pgurl, tc.replicationSlotName)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantstatus)
+		})
+	}
+}
+
+func TestStatusChecker_transformationRulesStatus(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name                 string
+		ruleValidatorBuilder func(context.Context, string) (ruleValidator, error)
+		config               *Config
+		wantStatus           *TransformationRulesStatus
+		wantErr              error
+	}{
+		{
+			name: "ok - valid transformation rules",
+			ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+				return func(rules []transformer.TableRules) (map[string]transformer.ColumnTransformers, error) {
+					return nil, nil
+				}, nil
+			},
+			config: &Config{
+				Processor: ProcessorConfig{
+					Transformer: &transformer.Config{
+						TransformerRules: []transformer.TableRules{},
+					},
+				},
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &TransformationRulesStatus{
+				Valid: true,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - no transformer configured",
+			ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+				return nil, errors.New("unexpected call to ruleValidatorBuilder")
+			},
+			config: &Config{
+				Processor: ProcessorConfig{
+					Transformer: nil,
+				},
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: nil,
+			wantErr:    nil,
+		},
+		{
+			name: "error - source postgres URL not provided",
+			ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+				return nil, errors.New("unexpected call to ruleValidatorBuilder")
+			},
+			config: &Config{
+				Processor: ProcessorConfig{
+					Transformer: &transformer.Config{
+						TransformerRules: []transformer.TableRules{},
+					},
+				},
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "",
+				}}},
+			},
+			wantStatus: &TransformationRulesStatus{
+				Valid:  false,
+				Errors: []string{fmt.Sprintf("cannot validate transformer rules: %s", sourceNotProvided)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - rule validation failure",
+			ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+				return func(rules []transformer.TableRules) (map[string]transformer.ColumnTransformers, error) {
+					return nil, errTest
+				}, nil
+			},
+			config: &Config{
+				Processor: ProcessorConfig{
+					Transformer: &transformer.Config{
+						TransformerRules: []transformer.TableRules{},
+					},
+				},
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: &TransformationRulesStatus{
+				Valid:  false,
+				Errors: []string{errTest.Error()},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - ruleValidatorBuilder failure",
+			ruleValidatorBuilder: func(ctx context.Context, pgURL string) (ruleValidator, error) {
+				return nil, errTest
+			},
+			config: &Config{
+				Processor: ProcessorConfig{
+					Transformer: &transformer.Config{
+						TransformerRules: []transformer.TableRules{},
+					},
+				},
+				Listener: ListenerConfig{Postgres: &PostgresListenerConfig{Replication: replicationpg.Config{
+					PostgresURL: "postgres://user:password@localhost:5432/db",
+				}}},
+			},
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.ruleValidatorBuilder = tc.ruleValidatorBuilder
+
+			status, err := sc.transformationRulesStatus(context.Background(), tc.config)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
+
+func TestStatusChecker_validateSchemaStatus(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name        string
+		connBuilder pglib.QuerierBuilder
+		pgurl       string
+
+		wantStatus *SchemaStatus
+		wantErr    error
+	}{
+		{
+			name: "ok - schema and schema_log table exist",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &SchemaStatus{
+				SchemaExists:         true,
+				SchemaLogTableExists: true,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - schema exists but schema_log table does not",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = false
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &SchemaStatus{
+				SchemaExists:         true,
+				SchemaLogTableExists: false,
+				Errors:               []string{noPgstreamSchemaLogTableErrMsg},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - schema does not exist",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = false
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &SchemaStatus{
+				SchemaExists:         false,
+				SchemaLogTableExists: false,
+				Errors:               []string{noPgstreamSchemaErrMsg},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - query failure when checking schema existence",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								return errTest
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			pgurl:      "postgres://user:password@localhost:5432/db",
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name: "error - query failure when checking schema_log table existence",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgstream')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 1)
+								exists, ok := dest[0].(*bool)
+								require.True(t, ok)
+								*exists = true
+								return nil
+							}}
+						case "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgstream' AND table_name = 'schema_log')":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								return errTest
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			pgurl:      "postgres://user:password@localhost:5432/db",
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.connBuilder = tc.connBuilder
+
+			status, err := sc.validateSchemaStatus(context.Background(), tc.pgurl)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
+
+func TestStatusChecker_validateMigrationStatus(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+	okMigrationVersion := uint(len(pgmigrations.AssetNames()) / 2)
+
+	tests := []struct {
+		name            string
+		migratorBuilder func(string) (migrator, error)
+		pgurl           string
+
+		wantStatus *MigrationStatus
+		wantErr    error
+	}{
+		{
+			name: "ok - valid migration status",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return okMigrationVersion, false, nil
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &MigrationStatus{
+				Version: okMigrationVersion,
+				Dirty:   false,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - pgstream schema doesn't exist",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return nil, errors.New("failed to open database: no schema")
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &MigrationStatus{
+				Errors: []string{noMigrationsTableErrMsg},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - dirty migration",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return okMigrationVersion, true, nil
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &MigrationStatus{
+				Version: okMigrationVersion,
+				Dirty:   true,
+				Errors:  []string{fmt.Sprintf("migration version %d is dirty", okMigrationVersion)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - mismatched migration files",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return 3, false, nil
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &MigrationStatus{
+				Version: 3,
+				Dirty:   false,
+				Errors:  []string{fmt.Sprintf("migration version (3) does not match the number of migration files (%d)", okMigrationVersion)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - multiple migration errors",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return 3, true, nil
+					},
+				}, nil
+			},
+			pgurl: "postgres://user:password@localhost:5432/db",
+			wantStatus: &MigrationStatus{
+				Version: 3,
+				Dirty:   true,
+				Errors:  []string{fmt.Sprintf("migration version (3) does not match the number of migration files (%d)", okMigrationVersion), "migration version 3 is dirty"},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - failed to create migrator",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return nil, errTest
+			},
+			pgurl:      "postgres://user:password@localhost:5432/db",
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+		{
+			name: "error - failed to get migration version",
+			migratorBuilder: func(pgURL string) (migrator, error) {
+				return &mockMigrator{
+					versionFn: func() (uint, bool, error) {
+						return 0, false, errTest
+					},
+				}, nil
+			},
+			pgurl:      "postgres://user:password@localhost:5432/db",
+			wantStatus: nil,
+			wantErr:    errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.migratorBuilder = tc.migratorBuilder
+
+			status, err := sc.validateMigrationStatus(tc.pgurl)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
+
+func TestStatusChecker_validateReplicationSlotStatus(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+	const testReplicationSlot = "pgstream_db_slot"
+	const testDB = "db"
+
+	tests := []struct {
+		name                string
+		connBuilder         pglib.QuerierBuilder
+		configParser        func(pgURL string) (*pgx.ConnConfig, error)
+		replicationSlotName string
+		pgurl               string
+
+		wantStatus *ReplicationSlotStatus
+		wantErr    error
+	}{
+		{
+			name: "ok - valid replication slot status",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							require.Len(t, args, 1)
+							require.Equal(t, args[0], testReplicationSlot)
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 3)
+
+								name, ok := dest[0].(*string)
+								require.True(t, ok)
+								*name = testReplicationSlot
+
+								plugin, ok := dest[1].(*string)
+								require.True(t, ok)
+								*plugin = wal2jsonPlugin
+
+								db, ok := dest[2].(*string)
+								require.True(t, ok)
+								*db = testDB
+
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			replicationSlotName: testReplicationSlot,
+			pgurl:               "postgres://user:password@localhost:5432/db",
+			wantStatus: &ReplicationSlotStatus{
+				Name:     testReplicationSlot,
+				Plugin:   wal2jsonPlugin,
+				Database: testDB,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - replication slot does not exist",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								return pglib.ErrNoRows
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			replicationSlotName: testReplicationSlot,
+			pgurl:               "postgres://user:password@localhost:5432/db",
+			wantStatus: &ReplicationSlotStatus{
+				Errors: []string{fmt.Sprintf("replication slot %s does not exist in the configured database", testReplicationSlot)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - replication slot on wrong database",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 3)
+
+								name, ok := dest[0].(*string)
+								require.True(t, ok)
+								*name = testReplicationSlot
+
+								plugin, ok := dest[1].(*string)
+								require.True(t, ok)
+								*plugin = wal2jsonPlugin
+
+								db, ok := dest[2].(*string)
+								require.True(t, ok)
+								*db = "wrong_db"
+
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			replicationSlotName: testReplicationSlot,
+			pgurl:               "postgres://user:password@localhost:5432/db",
+			wantStatus: &ReplicationSlotStatus{
+				Name:     testReplicationSlot,
+				Plugin:   wal2jsonPlugin,
+				Database: "wrong_db",
+				Errors:   []string{fmt.Sprintf("replication slot %s does not exist in the configured database", testReplicationSlot)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - replication slot using wrong plugin",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						switch sql {
+						case "SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_name = $1":
+							return &pgmocks.Row{ScanFn: func(dest ...any) error {
+								require.Len(t, dest, 3)
+
+								name, ok := dest[0].(*string)
+								require.True(t, ok)
+								*name = testReplicationSlot
+
+								plugin, ok := dest[1].(*string)
+								require.True(t, ok)
+								*plugin = "wrong_plugin"
+
+								db, ok := dest[2].(*string)
+								require.True(t, ok)
+								*db = testDB
+
+								return nil
+							}}
+						default:
+							return &pgmocks.Row{ScanFn: func(args ...any) error { return fmt.Errorf("unexpected query: %v", sql) }}
+						}
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			replicationSlotName: testReplicationSlot,
+			pgurl:               "postgres://user:password@localhost:5432/db",
+			wantStatus: &ReplicationSlotStatus{
+				Name:     testReplicationSlot,
+				Plugin:   "wrong_plugin",
+				Database: testDB,
+				Errors:   []string{fmt.Sprintf("replication slot %s is not using the wal2json plugin", testReplicationSlot)},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - query failure",
+			connBuilder: func(ctx context.Context, pgURL string) (pglib.Querier, error) {
+				return &pgmocks.Querier{
+					QueryRowFn: func(ctx context.Context, sql string, args ...any) pglib.Row {
+						return &pgmocks.Row{ScanFn: func(dest ...any) error {
+							return errTest
+						}}
+					},
+				}, nil
+			},
+			configParser: func(pgURL string) (*pgx.ConnConfig, error) {
+				return &pgx.ConnConfig{Config: pgconn.Config{Database: testDB}}, nil
+			},
+			replicationSlotName: testReplicationSlot,
+			pgurl:               "postgres://user:password@localhost:5432/db",
+			wantStatus:          nil,
+			wantErr:             errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			sc := NewStatusChecker()
+			sc.connBuilder = tc.connBuilder
+			sc.configParser = tc.configParser
+
+			status, err := sc.validateReplicationSlotStatus(context.Background(), tc.pgurl, tc.replicationSlotName)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, status, tc.wantStatus)
+		})
+	}
+}
diff --git a/pkg/stream/stream_status_test.go b/pkg/stream/stream_status_test.go
new file mode 100644
index 0000000..bf36544
--- /dev/null
+++ b/pkg/stream/stream_status_test.go
@@ -0,0 +1,331 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package stream
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestInitStatus_GetErrors(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		initStatus *InitStatus
+		wantErrors []string
+	}{
+		{
+			name: "all components valid",
+			initStatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         true,
+					SchemaLogTableExists: true,
+				},
+				Migration: &MigrationStatus{
+					Version: 5,
+					Dirty:   false,
+				},
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     "pgstream_db_slot",
+					Plugin:   "wal2json",
+					Database: "db",
+				},
+			},
+			wantErrors: []string{},
+		},
+		{
+			name: "schema errors",
+			initStatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         false,
+					SchemaLogTableExists: false,
+					Errors:               []string{noPgstreamSchemaErrMsg},
+				},
+			},
+			wantErrors: []string{noPgstreamSchemaErrMsg},
+		},
+		{
+			name: "migration errors",
+			initStatus: &InitStatus{
+				Migration: &MigrationStatus{
+					Version: 3,
+					Dirty:   true,
+					Errors:  []string{"migration version 3 is dirty"},
+				},
+			},
+			wantErrors: []string{"migration version 3 is dirty"},
+		},
+		{
+			name: "replication slot errors",
+			initStatus: &InitStatus{
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     "pgstream_db_slot",
+					Plugin:   "wrong_plugin",
+					Database: "db",
+					Errors:   []string{"replication slot pgstream_db_slot is not using the wal2json plugin"},
+				},
+			},
+			wantErrors: []string{"replication slot pgstream_db_slot is not using the wal2json plugin"},
+		},
+		{
+			name: "multiple errors",
+			initStatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         false,
+					SchemaLogTableExists: false,
+					Errors:               []string{noPgstreamSchemaErrMsg},
+				},
+				Migration: &MigrationStatus{
+					Version: 3,
+					Dirty:   true,
+					Errors:  []string{"migration version 3 is dirty"},
+				},
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     "pgstream_db_slot",
+					Plugin:   "wrong_plugin",
+					Database: "db",
+					Errors:   []string{"replication slot pgstream_db_slot is not using the wal2json plugin"},
+				},
+			},
+			wantErrors: []string{
+				noPgstreamSchemaErrMsg,
+				"migration version 3 is dirty",
+				"replication slot pgstream_db_slot is not using the wal2json plugin",
+			},
+		},
+		{
+			name:       "nil InitStatus",
+			initStatus: nil,
+			wantErrors: nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			errors := tc.initStatus.GetErrors()
+			require.Equal(t, tc.wantErrors, errors)
+		})
+	}
+}
+
+func TestStatus_PrettyPrint(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		status     *Status
+		wantOutput string
+	}{
+		{
+			name: "all components valid",
+			status: &Status{
+				Init: &InitStatus{
+					PgstreamSchema: &SchemaStatus{
+						SchemaExists:         true,
+						SchemaLogTableExists: true,
+					},
+					Migration: &MigrationStatus{
+						Version: 5,
+						Dirty:   false,
+					},
+					ReplicationSlot: &ReplicationSlotStatus{
+						Name:     "pgstream_db_slot",
+						Plugin:   "wal2json",
+						Database: "db",
+					},
+				},
+				Config: &ConfigStatus{
+					Valid: true,
+				},
+				Source: &SourceStatus{
+					Reachable: true,
+				},
+				TransformationRules: &TransformationRulesStatus{
+					Valid: true,
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Pgstream schema exists: true
+ - Pgstream schema_log table exists: true
+ - Migration current version: 5
+ - Migration status: success
+ - Replication slot name: pgstream_db_slot
+ - Replication slot plugin: wal2json
+ - Replication slot database: db
+Config status:
+ - Valid: true
+Transformation rules status:
+ - Valid: true
+Source status:
+ - Reachable: true`,
+		},
+		{
+			name: "components with errors",
+			status: &Status{
+				Init: &InitStatus{
+					PgstreamSchema: &SchemaStatus{
+						SchemaExists:         false,
+						SchemaLogTableExists: false,
+						Errors:               []string{"pgstream schema does not exist"},
+					},
+					Migration: &MigrationStatus{
+						Version: 3,
+						Dirty:   true,
+						Errors:  []string{"migration version 3 is dirty"},
+					},
+					ReplicationSlot: &ReplicationSlotStatus{
+						Name:     "pgstream_db_slot",
+						Plugin:   "wrong_plugin",
+						Database: "db",
+						Errors:   []string{"replication slot is not using wal2json plugin"},
+					},
+				},
+				Config: &ConfigStatus{
+					Valid:  false,
+					Errors: []string{"invalid configuration"},
+				},
+				Source: &SourceStatus{
+					Reachable: false,
+					Errors:    []string{"source not reachable"},
+				},
+				TransformationRules: &TransformationRulesStatus{
+					Valid:  false,
+					Errors: []string{"invalid transformation rules"},
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Pgstream schema exists: false
+ - Pgstream schema_log table exists: false
+ - Pgstream schema errors: [pgstream schema does not exist]
+ - Migration current version: 3
+ - Migration status: failed
+ - Migration errors: [migration version 3 is dirty]
+ - Replication slot name: pgstream_db_slot
+ - Replication slot plugin: wrong_plugin
+ - Replication slot database: db
+ - Replication slot errors: [replication slot is not using wal2json plugin]
+Config status:
+ - Valid: false
+ - Errors: [invalid configuration]
+Transformation rules status:
+ - Valid: false
+ - Errors: [invalid transformation rules]
+Source status:
+ - Reachable: false
+ - Errors: [source not reachable]`,
+		},
+		{
+			name:       "nil status",
+			status:     nil,
+			wantOutput: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			output := tc.status.PrettyPrint()
+			require.Equal(t, tc.wantOutput, output)
+		})
+	}
+}
+
+func TestInitStatus_PrettyPrint(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		initStatus *InitStatus
+		wantOutput string
+	}{
+		{
+			name: "all components valid",
+			initStatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         true,
+					SchemaLogTableExists: true,
+				},
+				Migration: &MigrationStatus{
+					Version: 5,
+					Dirty:   false,
+				},
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     "pgstream_db_slot",
+					Plugin:   "wal2json",
+					Database: "db",
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Pgstream schema exists: true
+ - Pgstream schema_log table exists: true
+ - Migration current version: 5
+ - Migration status: success
+ - Replication slot name: pgstream_db_slot
+ - Replication slot plugin: wal2json
+ - Replication slot database: db`,
+		},
+		{
+			name: "schema errors",
+			initStatus: &InitStatus{
+				PgstreamSchema: &SchemaStatus{
+					SchemaExists:         false,
+					SchemaLogTableExists: false,
+					Errors:               []string{noPgstreamSchemaErrMsg},
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Pgstream schema exists: false
+ - Pgstream schema_log table exists: false
+ - Pgstream schema errors: [pgstream schema does not exist in the configured postgres database]`,
+		},
+		{
+			name: "migration errors",
+			initStatus: &InitStatus{
+				Migration: &MigrationStatus{
+					Version: 3,
+					Dirty:   true,
+					Errors:  []string{"migration version 3 is dirty"},
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Migration current version: 3
+ - Migration status: failed
+ - Migration errors: [migration version 3 is dirty]`,
+		},
+		{
+			name: "replication slot errors",
+			initStatus: &InitStatus{
+				ReplicationSlot: &ReplicationSlotStatus{
+					Name:     "pgstream_db_slot",
+					Plugin:   "wrong_plugin",
+					Database: "db",
+					Errors:   []string{"replication slot pgstream_db_slot is not using the wal2json plugin"},
+				},
+			},
+			wantOutput: `Initialisation status:
+ - Replication slot name: pgstream_db_slot
+ - Replication slot plugin: wrong_plugin
+ - Replication slot database: db
+ - Replication slot errors: [replication slot pgstream_db_slot is not using the wal2json plugin]`,
+		},
+		{
+			name:       "nil InitStatus",
+			initStatus: nil,
+			wantOutput: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			output := tc.initStatus.PrettyPrint()
+			require.Equal(t, tc.wantOutput, output)
+		})
+	}
+}