Repository: xataio/pgstream
Pull request #256 (2025-04-08 10:24:57)

Add yaml config support
This PR adds support to provide the pgstream configuration as a single unified yaml file. In this file, the existing yaml file used to define the transformer rules is integrated, so that everything is configured in a single place.

The new yaml configuration aims to allow a simpler setup, helping configure correctly the increasing number of target/sources and modes.

The environment variables are still supported, and will be parsed as normal if provided. The only difference is that the transformation rules yaml file from the `PGSTREAM_TRANSFORMER_RULES_FILE` environment variable is now unmarshaled as part of the cmd directory, instead of being part of the wal transformer processor implementation. This aims to decouple the configuration specifics from the processor implementation itself.

The env and yaml configurations are kept separate, and only one or the other should be provided. This is done for backwards compatibility, since the environment variable names don't match those in the yaml file, and it might otherwise detract from the goal of having a simpler more readable configuration. If we need to support having both, we can try to align the naming so that the viper library can automatically handle environment variable overrides.

A config template with all possible options is added under `config_template.yaml` as well as sample configs for all existing configurations to match the `.env` existing files. 
https://github.com/xataio/pgstream/blob/use-yaml-config/config_template.yaml


Patch:
diff --git a/cli-definition.json b/cli-definition.json
index ee49ea1..8e6a903 100644
--- a/cli-definition.json
+++ b/cli-definition.json
@@ -33,7 +33,7 @@
     {
       "name": "config",
       "shorthand": "c",
-      "description": ".env config file to use if any",
+      "description": ".env or .yaml config file to use with pgstream if any",
       "default": ""
     },
     {
diff --git a/cmd/config/config.go b/cmd/config/config.go
new file mode 100644
index 0000000..4be9dfa
--- /dev/null
+++ b/cmd/config/config.go
@@ -0,0 +1,82 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"fmt"
+	"path/filepath"
+
+	"github.com/spf13/viper"
+	"github.com/xataio/pgstream/pkg/stream"
+)
+
+func Load() error {
+	return LoadFile(viper.GetString("config"))
+}
+
+func LoadFile(file string) error {
+	if file != "" {
+		viper.SetConfigFile(file)
+		viper.SetConfigType(filepath.Ext(file)[1:])
+		if err := viper.ReadInConfig(); err != nil {
+			return fmt.Errorf("reading config: %w", err)
+		}
+	}
+
+	transformerRulesFile := viper.GetString("PGSTREAM_TRANSFORMER_RULES_FILE")
+	if transformerRulesFile != "" {
+		viper.SetConfigFile(transformerRulesFile)
+		viper.SetConfigType(filepath.Ext(transformerRulesFile)[1:])
+		if err := viper.MergeInConfig(); err != nil {
+			return fmt.Errorf("reading transformer rules config: %w", err)
+		}
+	}
+	return nil
+}
+
+func PostgresURL() (url string) {
+	switch {
+	case viper.GetString("source.postgres.url") != "":
+		// yaml config
+		return viper.GetString("source.postgres.url")
+	case viper.GetString("PGSTREAM_POSTGRES_LISTENER_URL") != "":
+		// env config
+		return viper.GetString("PGSTREAM_POSTGRES_LISTENER_URL")
+	case viper.GetString("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL") != "":
+		// env config
+		return viper.GetString("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL")
+	default:
+		// CLI argument (with default value)
+		return viper.GetString("pgurl")
+	}
+}
+
+func ReplicationSlotName() string {
+	switch {
+	case viper.GetString("replication-slot") != "":
+		// CLI argument
+		return viper.GetString("replication-slot")
+	case viper.GetString("source.postgres.replication.replication_slot") != "":
+		// yaml config
+		return viper.GetString("source.postgres.replication.replication_slot")
+	case viper.GetString("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME") != "":
+		// env config
+		return viper.GetString("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME")
+	default:
+		return ""
+	}
+}
+
+func ParseStreamConfig() (*stream.Config, error) {
+	cfgFile := viper.GetViper().ConfigFileUsed()
+	switch ext := filepath.Ext(cfgFile); ext {
+	case ".yml", ".yaml":
+		yamlCfg := YAMLConfig{}
+		if err := viper.Unmarshal(&yamlCfg); err != nil {
+			return nil, err
+		}
+		return yamlCfg.toStreamConfig()
+	default:
+		return envConfigToStreamConfig()
+	}
+}
diff --git a/cmd/config.go b/cmd/config/config_env.go
similarity index 88%
rename from cmd/config.go
rename to cmd/config/config_env.go
index 3aae2d8..536787a 100644
--- a/cmd/config.go
+++ b/cmd/config/config_env.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package cmd
+package config
 
 import (
 	"fmt"
@@ -28,42 +28,15 @@ import (
 	pgreplication "github.com/xataio/pgstream/pkg/wal/replication/postgres"
 )
 
-func loadConfig() error {
-	cfgFile := viper.GetString("config")
-	if cfgFile != "" {
-		fmt.Printf("using config file: %s\n", cfgFile) //nolint:forbidigo //logger hasn't been configured yet
-		viper.SetConfigFile(cfgFile)
-		if err := viper.ReadInConfig(); err != nil {
-			return fmt.Errorf("error reading config: %w", err)
-		}
-	}
-	return nil
-}
-
-func pgURL() (url string) {
-	switch {
-	case viper.GetString("PGSTREAM_POSTGRES_LISTENER_URL") != "":
-		return viper.GetString("PGSTREAM_POSTGRES_LISTENER_URL")
-	case viper.GetString("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL") != "":
-		return viper.GetString("PGSTREAM_POSTGRES_SNAPSHOT_LISTENER_URL")
-	default:
-		return viper.GetString("pgurl")
-	}
-}
-
-func replicationSlotName() string {
-	replicationslot := viper.GetString("replication-slot")
-	if replicationslot != "" {
-		return replicationslot
+func envConfigToStreamConfig() (*stream.Config, error) {
+	processorCfg, err := parseProcessorConfig()
+	if err != nil {
+		return nil, err
 	}
-	return viper.GetString("PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME")
-}
-
-func parseStreamConfig() *stream.Config {
 	return &stream.Config{
 		Listener:  parseListenerConfig(),
-		Processor: parseProcessorConfig(),
-	}
+		Processor: processorCfg,
+	}, nil
 }
 
 // listener parsing
@@ -85,7 +58,7 @@ func parsePostgresListenerConfig() *stream.PostgresListenerConfig {
 	cfg := &stream.PostgresListenerConfig{
 		Replication: pgreplication.Config{
 			PostgresURL:         pgURL,
-			ReplicationSlotName: replicationSlotName(),
+			ReplicationSlotName: ReplicationSlotName(),
 		},
 	}
 
@@ -177,15 +150,19 @@ func parseKafkaCheckpointConfig() kafkacheckpoint.Config {
 
 // processor parsing
 
-func parseProcessorConfig() stream.ProcessorConfig {
+func parseProcessorConfig() (stream.ProcessorConfig, error) {
+	transformerCfg, err := parseTransformerConfig()
+	if err != nil {
+		return stream.ProcessorConfig{}, err
+	}
 	return stream.ProcessorConfig{
 		Kafka:       parseKafkaProcessorConfig(),
 		Search:      parseSearchProcessorConfig(),
 		Webhook:     parseWebhookProcessorConfig(),
 		Postgres:    parsePostgresProcessorConfig(),
 		Injector:    parseInjectorConfig(),
-		Transformer: parseTransformerConfig(),
-	}
+		Transformer: transformerCfg,
+	}, nil
 }
 
 func parseKafkaProcessorConfig() *stream.KafkaProcessorConfig {
@@ -235,6 +212,7 @@ func parseSearchProcessorConfig() *stream.SearchProcessorConfig {
 				MaxBatchSize:  viper.GetInt64("PGSTREAM_SEARCH_INDEXER_BATCH_SIZE"),
 				BatchTimeout:  viper.GetDuration("PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT"),
 				MaxQueueBytes: viper.GetInt64("PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES"),
+				MaxBatchBytes: viper.GetInt64("PGSTREAM_SEARCH_INDEXER_BATCH_BYTES"),
 			},
 		},
 		Store: store.Config{
@@ -341,14 +319,17 @@ func parseInjectorConfig() *injector.Config {
 	}
 }
 
-func parseTransformerConfig() *transformer.Config {
-	transformerRulesFile := viper.GetString("PGSTREAM_TRANSFORMER_RULES_FILE")
-	if transformerRulesFile == "" {
-		return nil
+func parseTransformerConfig() (*transformer.Config, error) {
+	if viper.GetString("PGSTREAM_TRANSFORMER_RULES_FILE") == "" {
+		return nil, nil
 	}
-	return &transformer.Config{
-		TransformerRulesFile: transformerRulesFile,
+	yamlConfig := struct {
+		Transformations TransformationsConfig `mapstructure:"transformations"`
+	}{}
+	if err := viper.Unmarshal(&yamlConfig); err != nil {
+		return nil, err
 	}
+	return yamlConfig.Transformations.parseTransformationConfig(), nil
 }
 
 func parseTLSConfig(prefix string) tls.Config {
diff --git a/cmd/config/config_yaml.go b/cmd/config/config_yaml.go
new file mode 100644
index 0000000..6ec960f
--- /dev/null
+++ b/cmd/config/config_yaml.go
@@ -0,0 +1,646 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/xataio/pgstream/pkg/backoff"
+	"github.com/xataio/pgstream/pkg/kafka"
+	pgschemalog "github.com/xataio/pgstream/pkg/schemalog/postgres"
+	pgsnapshotgenerator "github.com/xataio/pgstream/pkg/snapshot/generator/postgres/data"
+	"github.com/xataio/pgstream/pkg/snapshot/generator/postgres/schema/pgdumprestore"
+	"github.com/xataio/pgstream/pkg/stream"
+	"github.com/xataio/pgstream/pkg/tls"
+	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
+	"github.com/xataio/pgstream/pkg/wal/listener/snapshot/adapter"
+	snapshotbuilder "github.com/xataio/pgstream/pkg/wal/listener/snapshot/builder"
+	"github.com/xataio/pgstream/pkg/wal/processor/batch"
+	"github.com/xataio/pgstream/pkg/wal/processor/injector"
+	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
+	"github.com/xataio/pgstream/pkg/wal/processor/postgres"
+	"github.com/xataio/pgstream/pkg/wal/processor/search"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
+	"github.com/xataio/pgstream/pkg/wal/processor/transformer"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
+	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
+	pgreplication "github.com/xataio/pgstream/pkg/wal/replication/postgres"
+)
+
+type YAMLConfig struct {
+	Source    SourceConfig    `mapstructure:"source" yaml:"source"`
+	Target    TargetConfig    `mapstructure:"target" yaml:"target"`
+	Modifiers ModifiersConfig `mapstructure:"modifiers" yaml:"modifiers"`
+}
+
+type SourceConfig struct {
+	Postgres *PostgresConfig `mapstructure:"postgres" yaml:"postgres"`
+	Kafka    *KafkaConfig    `mapstructure:"kafka" yaml:"kafka"`
+}
+
+type TargetConfig struct {
+	Postgres *PostgresTargetConfig `mapstructure:"postgres" yaml:"postgres"`
+	Kafka    *KafkaTargetConfig    `mapstructure:"kafka" yaml:"kafka"`
+	Search   *SearchConfig         `mapstructure:"search" yaml:"search"`
+	Webhooks *WebhooksConfig       `mapstructure:"webhooks" yaml:"webhooks"`
+}
+
+type PostgresConfig struct {
+	URL         string             `mapstructure:"url" yaml:"url"`
+	Mode        string             `mapstructure:"mode" yaml:"mode"`
+	Snapshot    *SnapshotConfig    `mapstructure:"snapshot" yaml:"snapshot"`
+	Replication *ReplicationConfig `mapstructure:"replication" yaml:"replication"`
+}
+
+type SnapshotConfig struct {
+	Mode            string                `mapstructure:"mode" yaml:"mode"`
+	Tables          []string              `mapstructure:"tables" yaml:"tables"`
+	RecorderEnabled bool                  `mapstructure:"recorder_enabled" yaml:"recorder_enabled"`
+	SnapshotWorkers int                   `mapstructure:"snapshot_workers" yaml:"snapshot_workers"`
+	Data            *SnapshotDataConfig   `mapstructure:"data" yaml:"data"`
+	Schema          *SnapshotSchemaConfig `mapstructure:"schema" yaml:"schema"`
+}
+
+type SnapshotDataConfig struct {
+	SchemaWorkers int `mapstructure:"schema_workers" yaml:"schema_workers"`
+	TableWorkers  int `mapstructure:"table_workers" yaml:"table_workers"`
+	BatchPageSize int `mapstructure:"batch_page_size" yaml:"batch_page_size"`
+}
+
+type SnapshotSchemaConfig struct {
+	Mode            string                 `mapstructure:"mode" yaml:"mode"`
+	PgDumpPgRestore *PgDumpPgRestoreConfig `mapstructure:"pgdump_pgrestore" yaml:"pgdump_pgrestore"`
+}
+
+type PgDumpPgRestoreConfig struct {
+	CleanTargetDB bool `mapstructure:"clean_target_db" yaml:"clean_target_db"`
+}
+
+type ReplicationConfig struct {
+	ReplicationSlot string `mapstructure:"replication_slot" yaml:"replication_slot"`
+}
+
+type KafkaConfig struct {
+	Servers       []string            `mapstructure:"servers" yaml:"servers"`
+	Topic         TopicConfig         `mapstructure:"topic" yaml:"topic"`
+	ConsumerGroup ConsumerGroupConfig `mapstructure:"consumer_group" yaml:"consumer_group"`
+	TLS           *TLSConfig          `mapstructure:"tls" yaml:"tls"`
+	Backoff       *BackoffConfig      `mapstructure:"backoff" yaml:"backoff"`
+}
+
+type TopicConfig struct {
+	Name              string `mapstructure:"name" yaml:"name"`
+	Partitions        int    `mapstructure:"partitions" yaml:"partitions"`
+	ReplicationFactor int    `mapstructure:"replication_factor" yaml:"replication_factor"`
+	AutoCreate        bool   `mapstructure:"auto_create" yaml:"auto_create"`
+}
+
+type ConsumerGroupConfig struct {
+	ID          string `mapstructure:"id" yaml:"id"`
+	StartOffset string `mapstructure:"start_offset" yaml:"start_offset"`
+}
+
+type TLSConfig struct {
+	CACert     string `mapstructure:"ca_cert" yaml:"ca_cert"`
+	ClientCert string `mapstructure:"client_cert" yaml:"client_cert"`
+	ClientKey  string `mapstructure:"client_key" yaml:"client_key"`
+}
+
+type BackoffConfig struct {
+	Exponential *ExponentialBackoffConfig `mapstructure:"exponential" yaml:"exponential"`
+	Constant    *ConstantBackoffConfig    `mapstructure:"constant" yaml:"constant"`
+}
+
+type ExponentialBackoffConfig struct {
+	MaxRetries      int `mapstructure:"max_retries" yaml:"max_retries"`
+	InitialInterval int `mapstructure:"initial_interval" yaml:"initial_interval"`
+	MaxInterval     int `mapstructure:"max_interval" yaml:"max_interval"`
+}
+
+type ConstantBackoffConfig struct {
+	MaxRetries int `mapstructure:"max_retries" yaml:"max_retries"`
+	Interval   int `mapstructure:"interval" yaml:"interval"`
+}
+
+type PostgresTargetConfig struct {
+	URL              string       `mapstructure:"url" yaml:"url"`
+	Batch            *BatchConfig `mapstructure:"batch" yaml:"batch"`
+	DisableTriggers  bool         `mapstructure:"disable_triggers" yaml:"disable_triggers"`
+	OnConflictAction string       `mapstructure:"on_conflict_action" yaml:"on_conflict_action"`
+}
+
+type KafkaTargetConfig struct {
+	Servers []string         `mapstructure:"servers" yaml:"servers"`
+	Topic   KafkaTopicConfig `mapstructure:"topic" yaml:"topic"`
+	TLS     *TLSConfig       `mapstructure:"tls" yaml:"tls"`
+	Batch   *BatchConfig     `mapstructure:"batch" yaml:"batch"`
+}
+
+type KafkaTopicConfig struct {
+	Name              string `mapstructure:"name" yaml:"name"`
+	Partitions        int    `mapstructure:"partitions" yaml:"partitions"`
+	ReplicationFactor int    `mapstructure:"replication_factor" yaml:"replication_factor"`
+	AutoCreate        bool   `mapstructure:"auto_create" yaml:"auto_create"`
+}
+
+type SearchConfig struct {
+	Engine  string         `mapstructure:"engine" yaml:"engine"`
+	URL     string         `mapstructure:"url" yaml:"url"`
+	Batch   *BatchConfig   `mapstructure:"batch" yaml:"batch"`
+	Backoff *BackoffConfig `mapstructure:"backoff" yaml:"backoff"`
+}
+
+type BatchConfig struct {
+	Timeout       int `mapstructure:"timeout" yaml:"timeout"`
+	Size          int `mapstructure:"size" yaml:"size"`
+	MaxBytes      int `mapstructure:"max_bytes" yaml:"max_bytes"`
+	MaxQueueBytes int `mapstructure:"max_queue_bytes" yaml:"max_queue_bytes"`
+}
+
+type WebhooksConfig struct {
+	Subscriptions WebhookSubscriptionsConfig `mapstructure:"subscriptions" yaml:"subscriptions"`
+	Notifier      WebhookNotifierConfig      `mapstructure:"notifier" yaml:"notifier"`
+}
+
+type WebhookSubscriptionsConfig struct {
+	Store  WebhookStoreConfig  `mapstructure:"store" yaml:"store"`
+	Server WebhookServerConfig `mapstructure:"server" yaml:"server"`
+}
+
+type WebhookStoreConfig struct {
+	URL   string              `mapstructure:"url" yaml:"url"`
+	Cache *WebhookCacheConfig `mapstructure:"cache" yaml:"cache"`
+}
+
+type WebhookCacheConfig struct {
+	Enabled         bool `mapstructure:"enabled" yaml:"enabled"`
+	RefreshInterval int  `mapstructure:"refresh_interval" yaml:"refresh_interval"`
+}
+
+type WebhookServerConfig struct {
+	Address      string `mapstructure:"address" yaml:"address"`
+	ReadTimeout  int    `mapstructure:"read_timeout" yaml:"read_timeout"`
+	WriteTimeout int    `mapstructure:"write_timeout" yaml:"write_timeout"`
+}
+
+type WebhookNotifierConfig struct {
+	WorkerCount   int `mapstructure:"worker_count" yaml:"worker_count"`
+	ClientTimeout int `mapstructure:"client_timeout" yaml:"client_timeout"`
+}
+
+type ModifiersConfig struct {
+	Injector        *InjectorConfig       `mapstructure:"injector" yaml:"injector"`
+	Transformations TransformationsConfig `mapstructure:"transformations" yaml:"transformations"`
+}
+
+type InjectorConfig struct {
+	Enabled      bool   `mapstructure:"enabled" yaml:"enabled"`
+	SchemalogURL string `mapstructure:"schemalog_url" yaml:"schemalog_url"`
+}
+
+type TransformationsConfig []TransformationConfig
+
+type TransformationConfig struct {
+	Schema      string                              `mapstructure:"schema" yaml:"schema"`
+	Table       string                              `mapstructure:"table" yaml:"table"`
+	ColumnRules map[string]ColumnTransformersConfig `mapstructure:"column_transformers" yaml:"column_transformers"`
+}
+
+type ColumnTransformersConfig struct {
+	Name              string         `mapstructure:"name" yaml:"name"`
+	Parameters        map[string]any `mapstructure:"parameters" yaml:"parameters"`
+	DynamicParameters map[string]any `mapstructure:"dynamic_parameters" yaml:"dynamic_parameters"`
+}
+
+// postgres source modes
+const (
+	replicationMode            = "replication"
+	snapshotMode               = "snapshot"
+	snapshotAndReplicationMode = "snapshot_and_replication"
+)
+
+// snapshot modes
+const (
+	fullSnapshotMode   = "full"
+	dataSnapshotMode   = "data"
+	schemaSnapshotMode = "schema"
+)
+
+// schema snapshot modes
+const (
+	pgdumprestoreSchemaMode = "pgdump_pgrestore"
+	schemalogSchemaMode     = "schemalog"
+)
+
+// search engines
+const (
+	elasticsearchEngine = "elasticsearch"
+	opensearchEngine    = "opensearch"
+)
+
+var (
+	errUnsupportedSchemaSnapshotMode = errors.New("unsupported schema snapshot mode, must be one of 'pgdump_pgrestore' or 'schemalog'")
+	errUnsupportedSnapshotMode       = errors.New("unsupported snapshot mode, must be one of 'full', 'schema' or 'data'")
+	errUnsupportedPostgresSourceMode = errors.New("unsupported postgres source mode, must be one of 'replication', 'snapshot' or 'snapshot_and_replication'")
+	errUnsupportedSearchEngine       = errors.New("unsupported search engine, must be one of 'opensearch' or 'elasticsearch'")
+	errInvalidPgdumpPgrestoreConfig  = errors.New("pgdump_pgrestore snapshot mode requires target postgres config")
+	errInvalidInjectorConfig         = errors.New("injector config can't infer schemalog url from source postgres url, schemalog_url must be provided")
+)
+
+func (c *YAMLConfig) toStreamConfig() (*stream.Config, error) {
+	listener, err := c.parseListenerConfig()
+	if err != nil {
+		return nil, fmt.Errorf("parsing source config: %w", err)
+	}
+
+	processor, err := c.parseProcessorConfig()
+	if err != nil {
+		return nil, fmt.Errorf("parsing target and modifier config: %w", err)
+	}
+	return &stream.Config{
+		Listener:  listener,
+		Processor: processor,
+	}, nil
+}
+
+func (c *YAMLConfig) parseListenerConfig() (stream.ListenerConfig, error) {
+	pgListener, err := c.parsePostgresListenerConfig()
+	if err != nil {
+		return stream.ListenerConfig{}, fmt.Errorf("parsing postgres listener config: %w", err)
+	}
+
+	return stream.ListenerConfig{
+		Postgres: pgListener,
+		Kafka:    c.Source.Kafka.parseKafkaListenerConfig(),
+	}, nil
+}
+
+func (c *YAMLConfig) parseProcessorConfig() (stream.ProcessorConfig, error) {
+	streamCfg := stream.ProcessorConfig{
+		Kafka:       c.parseKafkaProcessorConfig(),
+		Postgres:    c.parsePostgresProcessorConfig(),
+		Webhook:     c.parseWebhookProcessorConfig(),
+		Transformer: c.parseTransformationConfig(),
+	}
+
+	var err error
+	streamCfg.Injector, err = c.parseInjectorConfig()
+	if err != nil {
+		return stream.ProcessorConfig{}, err
+	}
+
+	streamCfg.Search, err = c.parseSearchProcessorConfig()
+	if err != nil {
+		return stream.ProcessorConfig{}, err
+	}
+
+	return streamCfg, nil
+}
+
+func (c *YAMLConfig) parsePostgresListenerConfig() (*stream.PostgresListenerConfig, error) {
+	if c.Source.Postgres == nil {
+		return nil, nil
+	}
+
+	streamCfg := &stream.PostgresListenerConfig{
+		Replication: pgreplication.Config{
+			PostgresURL: c.Source.Postgres.URL,
+		},
+	}
+
+	switch c.Source.Postgres.Mode {
+	case replicationMode, snapshotMode, snapshotAndReplicationMode:
+	default:
+		return nil, errUnsupportedPostgresSourceMode
+	}
+
+	if c.Source.Postgres.Mode == replicationMode || c.Source.Postgres.Mode == snapshotAndReplicationMode {
+		replicationSlotName := ReplicationSlotName()
+		if c.Source.Postgres.Replication != nil {
+			replicationSlotName = c.Source.Postgres.Replication.ReplicationSlot
+		}
+		streamCfg.Replication.ReplicationSlotName = replicationSlotName
+	}
+
+	if c.Source.Postgres.Mode == snapshotMode || c.Source.Postgres.Mode == snapshotAndReplicationMode {
+		var err error
+		streamCfg.Snapshot, err = c.parseSnapshotConfig()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return streamCfg, nil
+}
+
+func (c *YAMLConfig) parseSnapshotConfig() (*snapshotbuilder.SnapshotListenerConfig, error) {
+	snapshotConfig := c.Source.Postgres.Snapshot
+	if snapshotConfig == nil {
+		return nil, nil
+	}
+
+	streamCfg := &snapshotbuilder.SnapshotListenerConfig{
+		Adapter: adapter.SnapshotConfig{
+			Tables:          snapshotConfig.Tables,
+			SnapshotWorkers: uint(snapshotConfig.SnapshotWorkers),
+		},
+	}
+
+	switch snapshotConfig.Mode {
+	case fullSnapshotMode, dataSnapshotMode, schemaSnapshotMode:
+	default:
+		return nil, errUnsupportedSnapshotMode
+	}
+
+	if snapshotConfig.Mode == fullSnapshotMode || snapshotConfig.Mode == dataSnapshotMode {
+		streamCfg.Generator = pgsnapshotgenerator.Config{
+			URL:           c.Source.Postgres.URL,
+			BatchPageSize: uint(snapshotConfig.Data.BatchPageSize),
+			SchemaWorkers: uint(snapshotConfig.Data.SchemaWorkers),
+			TableWorkers:  uint(snapshotConfig.Data.TableWorkers),
+		}
+	}
+
+	if snapshotConfig.Mode == fullSnapshotMode || snapshotConfig.Mode == schemaSnapshotMode {
+		var err error
+		streamCfg.Schema, err = c.parseSchemaSnapshotConfig()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if snapshotConfig.RecorderEnabled {
+		streamCfg.SnapshotStoreURL = c.Source.Postgres.URL
+	}
+
+	return streamCfg, nil
+}
+
+func (c *YAMLConfig) parseSchemaSnapshotConfig() (snapshotbuilder.SchemaSnapshotConfig, error) {
+	schemaSnapshotCfg := c.Source.Postgres.Snapshot.Schema
+	if schemaSnapshotCfg == nil {
+		return snapshotbuilder.SchemaSnapshotConfig{}, nil
+	}
+
+	switch schemaSnapshotCfg.Mode {
+	case schemalogSchemaMode:
+		return snapshotbuilder.SchemaSnapshotConfig{
+			SchemaLogStore: &pgschemalog.Config{
+				URL: c.Source.Postgres.URL,
+			},
+		}, nil
+	case pgdumprestoreSchemaMode:
+		if c.Target.Postgres == nil {
+			return snapshotbuilder.SchemaSnapshotConfig{}, errInvalidPgdumpPgrestoreConfig
+		}
+		streamSchemaCfg := snapshotbuilder.SchemaSnapshotConfig{
+			DumpRestore: &pgdumprestore.Config{
+				SourcePGURL: c.Source.Postgres.URL,
+				TargetPGURL: c.Target.Postgres.URL,
+			},
+		}
+
+		if schemaSnapshotCfg.PgDumpPgRestore != nil {
+			streamSchemaCfg.DumpRestore.CleanTargetDB = schemaSnapshotCfg.PgDumpPgRestore.CleanTargetDB
+		}
+
+		return streamSchemaCfg, nil
+	default:
+		return snapshotbuilder.SchemaSnapshotConfig{}, errUnsupportedSchemaSnapshotMode
+	}
+}
+
+func (c *YAMLConfig) parseInjectorConfig() (*injector.Config, error) {
+	if c.Modifiers.Injector == nil || !c.Modifiers.Injector.Enabled {
+		return nil, nil
+	}
+
+	url := c.Modifiers.Injector.SchemalogURL
+	if url == "" {
+		if c.Source.Postgres == nil || c.Source.Postgres.URL == "" {
+			return nil, errInvalidInjectorConfig
+		}
+		url = c.Source.Postgres.URL
+	}
+
+	return &injector.Config{
+		Store: pgschemalog.Config{
+			URL: url,
+		},
+	}, nil
+}
+
+func (c *YAMLConfig) parseKafkaProcessorConfig() *stream.KafkaProcessorConfig {
+	if c.Target.Kafka == nil {
+		return nil
+	}
+
+	return &stream.KafkaProcessorConfig{
+		Writer: &kafkaprocessor.Config{
+			Kafka: kafka.ConnConfig{
+				Servers: c.Target.Kafka.Servers,
+				Topic: kafka.TopicConfig{
+					Name:              c.Target.Kafka.Topic.Name,
+					NumPartitions:     c.Target.Kafka.Topic.Partitions,
+					ReplicationFactor: c.Target.Kafka.Topic.ReplicationFactor,
+					AutoCreate:        c.Target.Kafka.Topic.AutoCreate,
+				},
+				TLS: c.Target.Kafka.TLS.parseTLSConfig(),
+			},
+			Batch: c.Target.Kafka.Batch.parseBatchConfig(),
+		},
+	}
+}
+
+func (c *YAMLConfig) parsePostgresProcessorConfig() *stream.PostgresProcessorConfig {
+	if c.Target.Postgres == nil {
+		return nil
+	}
+
+	return &stream.PostgresProcessorConfig{
+		BatchWriter: postgres.Config{
+			URL:         c.Target.Postgres.URL,
+			BatchConfig: c.Target.Postgres.Batch.parseBatchConfig(),
+			SchemaLogStore: pgschemalog.Config{
+				URL: c.Source.Postgres.URL,
+			},
+			DisableTriggers:  c.Target.Postgres.DisableTriggers,
+			OnConflictAction: c.Target.Postgres.OnConflictAction,
+		},
+	}
+}
+
+func (c *YAMLConfig) parseSearchProcessorConfig() (*stream.SearchProcessorConfig, error) {
+	if c.Target.Search == nil {
+		return nil, nil
+	}
+
+	storeCfg := store.Config{}
+
+	switch c.Target.Search.Engine {
+	case elasticsearchEngine:
+		storeCfg.ElasticsearchURL = c.Target.Search.URL
+	case opensearchEngine:
+		storeCfg.OpenSearchURL = c.Target.Search.URL
+	default:
+		return nil, errUnsupportedSearchEngine
+	}
+
+	return &stream.SearchProcessorConfig{
+		Store: storeCfg,
+		Indexer: search.IndexerConfig{
+			Batch: c.Target.Search.Batch.parseBatchConfig(),
+		},
+		Retrier: search.StoreRetryConfig{
+			Backoff: c.Target.Search.Backoff.parseBackoffConfig(),
+		},
+	}, nil
+}
+
+func (c *YAMLConfig) parseWebhookProcessorConfig() *stream.WebhookProcessorConfig {
+	if c.Target.Webhooks == nil {
+		return nil
+	}
+	streamCfg := &stream.WebhookProcessorConfig{
+		SubscriptionStore: stream.WebhookSubscriptionStoreConfig{
+			URL: c.Target.Webhooks.Subscriptions.Store.URL,
+		},
+		Notifier: notifier.Config{
+			URLWorkerCount: uint(c.Target.Webhooks.Notifier.WorkerCount),
+			ClientTimeout:  time.Duration(c.Target.Webhooks.Notifier.ClientTimeout) * time.Millisecond,
+		},
+		SubscriptionServer: server.Config{
+			Address:      c.Target.Webhooks.Subscriptions.Server.Address,
+			ReadTimeout:  time.Duration(c.Target.Webhooks.Subscriptions.Server.ReadTimeout) * time.Second,
+			WriteTimeout: time.Duration(c.Target.Webhooks.Subscriptions.Server.WriteTimeout) * time.Second,
+		},
+	}
+
+	if c.Target.Webhooks.Subscriptions.Store.Cache != nil {
+		streamCfg.SubscriptionStore.CacheEnabled = c.Target.Webhooks.Subscriptions.Store.Cache.Enabled
+		streamCfg.SubscriptionStore.CacheRefreshInterval = time.Duration(c.Target.Webhooks.Subscriptions.Store.Cache.RefreshInterval) * time.Second
+	}
+
+	return streamCfg
+}
+
+func (c *YAMLConfig) parseTransformationConfig() *transformer.Config {
+	if c.Modifiers.Transformations == nil {
+		return nil
+	}
+
+	return c.Modifiers.Transformations.parseTransformationConfig()
+}
+
+func (c TransformationsConfig) parseTransformationConfig() *transformer.Config {
+	if len(c) == 0 {
+		return nil
+	}
+	rules := make([]transformer.TableRules, 0, len(c))
+	for _, t := range c {
+		columnRules := make(map[string]transformer.TransformerRules, len(t.ColumnRules))
+		for column, cr := range t.ColumnRules {
+			columnRules[column] = transformer.TransformerRules{
+				Name:              cr.Name,
+				Parameters:        cr.Parameters,
+				DynamicParameters: cr.DynamicParameters,
+			}
+		}
+		rules = append(rules, transformer.TableRules{
+			Schema:      t.Schema,
+			Table:       t.Table,
+			ColumnRules: columnRules,
+		})
+	}
+
+	return &transformer.Config{
+		TransformerRules: rules,
+	}
+}
+
+func (c *KafkaConfig) parseKafkaListenerConfig() *stream.KafkaListenerConfig {
+	if c == nil {
+		return nil
+	}
+
+	return &stream.KafkaListenerConfig{
+		Reader: c.parseKafkaReaderConfig(),
+		Checkpointer: kafkacheckpoint.Config{
+			CommitBackoff: c.Backoff.parseBackoffConfig(),
+		},
+	}
+}
+
+func (c *KafkaConfig) parseKafkaReaderConfig() kafka.ReaderConfig {
+	if c == nil {
+		return kafka.ReaderConfig{}
+	}
+	return kafka.ReaderConfig{
+		Conn: kafka.ConnConfig{
+			Servers: c.Servers,
+			Topic: kafka.TopicConfig{
+				Name:              c.Topic.Name,
+				NumPartitions:     c.Topic.Partitions,
+				ReplicationFactor: c.Topic.ReplicationFactor,
+				AutoCreate:        c.Topic.AutoCreate,
+			},
+			TLS: c.TLS.parseTLSConfig(),
+		},
+		ConsumerGroupID:          c.ConsumerGroup.ID,
+		ConsumerGroupStartOffset: c.ConsumerGroup.StartOffset,
+	}
+}
+
+func (t *TLSConfig) parseTLSConfig() tls.Config {
+	if t == nil {
+		return tls.Config{Enabled: false}
+	}
+	return tls.Config{
+		Enabled:        true,
+		CaCertFile:     t.CACert,
+		ClientCertFile: t.ClientCert,
+		ClientKeyFile:  t.ClientKey,
+	}
+}
+
+func (bo *BackoffConfig) parseBackoffConfig() backoff.Config {
+	return backoff.Config{
+		Exponential: bo.parseExponentialBackoffConfig(),
+		Constant:    bo.parseConstantBackoffConfig(),
+	}
+}
+
+func (bo *BackoffConfig) parseExponentialBackoffConfig() *backoff.ExponentialConfig {
+	if bo.Exponential == nil {
+		return nil
+	}
+	return &backoff.ExponentialConfig{
+		InitialInterval: time.Duration(bo.Exponential.InitialInterval) * time.Millisecond,
+		MaxInterval:     time.Duration(bo.Exponential.MaxInterval) * time.Millisecond,
+		MaxRetries:      uint(bo.Exponential.MaxRetries),
+	}
+}
+
+func (bo *BackoffConfig) parseConstantBackoffConfig() *backoff.ConstantConfig {
+	if bo.Constant == nil {
+		return nil
+	}
+	return &backoff.ConstantConfig{
+		Interval:   time.Duration(bo.Constant.Interval) * time.Millisecond,
+		MaxRetries: uint(bo.Constant.MaxRetries),
+	}
+}
+
+func (bc *BatchConfig) parseBatchConfig() batch.Config {
+	if bc == nil {
+		return batch.Config{}
+	}
+	return batch.Config{
+		BatchTimeout:  time.Duration(bc.Timeout) * time.Millisecond,
+		MaxBatchBytes: int64(bc.MaxBytes),
+		MaxQueueBytes: int64(bc.MaxQueueBytes),
+		MaxBatchSize:  int64(bc.Size),
+	}
+}
diff --git a/cmd/init_cmd.go b/cmd/init_cmd.go
index 473d7cb..7fa753f 100644
--- a/cmd/init_cmd.go
+++ b/cmd/init_cmd.go
@@ -5,6 +5,7 @@ package cmd
 import (
 	"context"
 
+	"github.com/xataio/pgstream/cmd/config"
 	"github.com/xataio/pgstream/pkg/stream"
 
 	_ "github.com/golang-migrate/migrate/v4/database/postgres"
@@ -18,7 +19,7 @@ var initCmd = &cobra.Command{
 	RunE: func(cmd *cobra.Command, args []string) error {
 		sp, _ := pterm.DefaultSpinner.WithText("initialising pgstream...").Start()
 
-		if err := stream.Init(context.Background(), pgURL(), replicationSlotName()); err != nil {
+		if err := stream.Init(context.Background(), config.PostgresURL(), config.ReplicationSlotName()); err != nil {
 			sp.Fail(err.Error())
 			return err
 		}
@@ -34,7 +35,7 @@ var tearDownCmd = &cobra.Command{
 	RunE: func(cmd *cobra.Command, args []string) error {
 		sp, _ := pterm.DefaultSpinner.WithText("tearing down pgstream...").Start()
 
-		if err := stream.TearDown(context.Background(), pgURL(), replicationSlotName()); err != nil {
+		if err := stream.TearDown(context.Background(), config.PostgresURL(), config.ReplicationSlotName()); err != nil {
 			sp.Fail(err.Error())
 			return err
 		}
diff --git a/cmd/root_cmd.go b/cmd/root_cmd.go
index 8e46959..fe22796 100644
--- a/cmd/root_cmd.go
+++ b/cmd/root_cmd.go
@@ -10,6 +10,8 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+
+	"github.com/xataio/pgstream/cmd/config"
 )
 
 // Version is the pgstream version
@@ -23,7 +25,7 @@ func Prepare() *cobra.Command {
 		SilenceUsage: true,
 		Version:      Version,
 		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
-			return loadConfig()
+			return config.Load()
 		},
 	}
 
@@ -32,7 +34,7 @@ func Prepare() *cobra.Command {
 
 	rootCmd.PersistentFlags().String("pgurl", "postgres://postgres:postgres@localhost?sslmode=disable", "Postgres URL")
 	rootCmd.PersistentFlags().String("replication-slot", "", "Name of the postgres replication slot to be created")
-	rootCmd.PersistentFlags().StringP("config", "c", "", ".env config file to use if any")
+	rootCmd.PersistentFlags().StringP("config", "c", "", ".env or .yaml config file to use with pgstream if any")
 	rootCmd.PersistentFlags().String("log-level", "debug", "log level for the application. One of trace, debug, info, warn, error, fatal, panic")
 
 	viper.BindPFlag("pgurl", rootCmd.PersistentFlags().Lookup("pgurl"))
diff --git a/cmd/start_cmd.go b/cmd/start_cmd.go
index 6b55d70..2b56081 100644
--- a/cmd/start_cmd.go
+++ b/cmd/start_cmd.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"github.com/xataio/pgstream/cmd/config"
 	"github.com/xataio/pgstream/internal/log/zerolog"
 	"github.com/xataio/pgstream/pkg/stream"
 )
@@ -22,5 +23,10 @@ func run(ctx context.Context) error {
 		LogLevel: viper.GetString("PGSTREAM_LOG_LEVEL"),
 	})
 	zerolog.SetGlobalLogger(logger)
-	return stream.Run(ctx, zerolog.NewStdLogger(logger), parseStreamConfig(), nil)
+
+	streamConfig, err := config.ParseStreamConfig()
+	if err != nil {
+		return err
+	}
+	return stream.Run(ctx, zerolog.NewStdLogger(logger), streamConfig, nil)
 }
diff --git a/config_template.yaml b/config_template.yaml
new file mode 100644
index 0000000..7bf8e41
--- /dev/null
+++ b/config_template.yaml
@@ -0,0 +1,110 @@
+source:
+  postgres:
+    url: "postgresql://user:password@localhost:5432/mydatabase"
+    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["test", "test_schema.test", "another_schema.*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      snapshot_workers: 4 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: pgdump_pgrestore # options are pgdump_pgrestore or schemalog
+        pgdump_pgrestore:
+          clean_target_db: true # whether to clean the target database before restoring
+    replication: # when mode is replication or snapshot_and_replication
+      replication_slot: "pgstream_mydatabase_slot"
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "mytopic"
+    consumer_group:
+      id: "mygroup"
+      start_offset: "earliest" # options are earliest or latest
+    tls:
+      ca_cert: "/path/to/ca.crt" # path to CA certificate
+      client_cert: "/path/to/client.crt" # path to client certificate
+      client_key: "/path/to/client.key" # path to client key
+    backoff: # one of exponential or constant
+      exponential:
+        max_retries: 5 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+      constant:
+        max_retries: 5 # maximum number of retries
+        interval: 1000 # interval in milliseconds
+
+target:
+  postgres:
+    url: "postgresql://user:password@localhost:5432/mytargetdatabase"
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+    disable_triggers: false # whether to disable triggers on the target database
+    on_conflict_action: "nothing" # options are update, nothing or error
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "mytopic" # name of the Kafka topic
+      partitions: 1 # number of partitions for the topic
+      replication_factor: 1 # replication factor for the topic
+      auto_create: true # whether to automatically create the topic if it doesn't exist
+    tls:
+      ca_cert: "/path/to/ca.crt" # path to CA certificate
+      client_cert: "/path/to/client.crt" # path to client certificate
+      client_key: "/path/to/client.key" # path to client key
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+  search:
+    engine: "elasticsearch" # options are elasticsearch or opensearch
+    url: "http://localhost:9200" # URL of the search engine
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+    backoff: # one of exponential or constant
+      exponential:
+        max_retries: 5 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+      constant:
+        max_retries: 5 # maximum number of retries
+        interval: 1000 # interval in milliseconds
+  webhooks:
+    subscriptions:
+      store:
+        url: "postgresql://user:password@localhost:5432/mydatabase" # URL of the database where the webhook subscriptions are stored
+        cache:
+          enabled: true # whether to enable caching for the subscription store
+          refresh_interval: 60 # interval in seconds to refresh the cache
+      server:
+        address: "localhost:9090" # address of the subscription server
+        read_timeout: 60 # read timeout in seconds
+        write_timeout: 60 # write timeout in seconds
+    notifier:
+      worker_count: 4 # number of notifications to be processed in parallel
+      client_timeout: 1000 # timeout for the webhook client in milliseconds
+
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
+    schemalog_url: "postgresql://user:password@localhost:5432/mydatabase" # URL of the schemalog database, if different from the source database
+  transformations:
+  - schema: public
+    table: test
+    column_transformers:
+      name:
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex
diff --git a/kafka2os.yaml b/kafka2os.yaml
new file mode 100644
index 0000000..0bcb215
--- /dev/null
+++ b/kafka2os.yaml
@@ -0,0 +1,31 @@
+source:
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "pgstream"
+    consumer_group:
+      id: "pgstream-consumer-group"
+      start_offset: "earliest" # options are earliest or latest
+    backoff: # one of exponential or constant
+      exponential:
+        max_retries: 60 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+
+target:
+  search:
+    engine: "opensearch" # options are elasticsearch or opensearch
+    url: "http://admin:admin@localhost:9200" # URL of the search engine
+    batch:
+      timeout: 5000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+    backoff: # one of exponential or constant
+      exponential:
+        max_retries: 0 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
+    schemalog_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the schemalog database, if different from the source database
diff --git a/pg2kafka.yaml b/pg2kafka.yaml
new file mode 100644
index 0000000..b1c8ce1
--- /dev/null
+++ b/pg2kafka.yaml
@@ -0,0 +1,32 @@
+source:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
+    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      snapshot_workers: 1 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: schemalog # options are pgdump_pgrestore or schemalog
+
+target:
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "pgstream" # name of the Kafka topic
+      partitions: 1 # number of partitions for the topic
+      replication_factor: 1 # replication factor for the topic
+      auto_create: true # whether to automatically create the topic if it doesn't exist
+    batch:
+      timeout: 2000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
diff --git a/pg2os.yaml b/pg2os.yaml
new file mode 100644
index 0000000..ea201f5
--- /dev/null
+++ b/pg2os.yaml
@@ -0,0 +1,33 @@
+source:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
+    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      snapshot_workers: 1 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: schemalog # options are pgdump_pgrestore or schemalog
+
+target:
+  search:
+    engine: "opensearch" # options are elasticsearch or opensearch
+    url: "http://admin:admin@localhost:9200" # URL of the search engine
+    batch:
+      timeout: 5000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+    backoff: # one of exponential or constant
+      exponential:
+        max_retries: 0 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
diff --git a/pg2pg.yaml b/pg2pg.yaml
new file mode 100644
index 0000000..0f93359
--- /dev/null
+++ b/pg2pg.yaml
@@ -0,0 +1,39 @@
+source:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
+    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      snapshot_workers: 1 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: pgdump_pgrestore # options are pgdump_pgrestore or schemalog
+        pgdump_pgrestore:
+          clean_target_db: true # whether to clean the target database before restoring
+
+target:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:7654?sslmode=disable"
+    batch:
+      timeout: 5000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+    disable_triggers: false # whether to disable triggers on the target database
+    on_conflict_action: "nothing" # options are update, nothing or error
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
+  transformations:
+  - schema: public
+    table: test
+    column_transformers:
+      name:
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex
diff --git a/pg2webhook.yaml b/pg2webhook.yaml
new file mode 100644
index 0000000..d849b4a
--- /dev/null
+++ b/pg2webhook.yaml
@@ -0,0 +1,24 @@
+source:
+  postgres:
+    url: "postgres://postgres:postgres@localhost:5432?sslmode=disable"
+    mode: replication # options are replication, snapshot or snapshot_and_replication
+
+target:
+  webhooks:
+    subscriptions:
+      store:
+        url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the database where the webhook subscriptions are stored
+        cache:
+          enabled: false # whether to enable caching for the subscription store
+      server:
+        address: "localhost:9090" # address of the subscription server
+        read_timeout: 60 # read timeout in seconds
+        write_timeout: 60 # write timeout in seconds
+    notifier:
+      worker_count: 4 # number of notifications to be processed in parallel
+      client_timeout: 10000 # timeout for the webhook client in milliseconds
+
+modifiers:
+   injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
+    schemalog_url: "postgres://postgres:postgres@localhost:5432?sslmode=disable" # URL of the schemalog database, if different from the source database
diff --git a/pkg/stream/config.go b/pkg/stream/config.go
index b9784dc..8736a15 100644
--- a/pkg/stream/config.go
+++ b/pkg/stream/config.go
@@ -4,6 +4,7 @@ package stream
 
 import (
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/xataio/pgstream/pkg/kafka"
@@ -77,13 +78,60 @@ type WebhookSubscriptionStoreConfig struct {
 }
 
 func (c *Config) IsValid() error {
-	if c.Listener.Kafka == nil && c.Listener.Postgres == nil && c.Listener.Snapshot == nil {
+	if err := c.Listener.IsValid(); err != nil {
+		return err
+	}
+
+	return c.Processor.IsValid()
+}
+
+func (c *ListenerConfig) IsValid() error {
+	listenerCount := 0
+	if c.Kafka != nil {
+		listenerCount++
+	}
+	if c.Postgres != nil {
+		listenerCount++
+	}
+	if c.Snapshot != nil {
+		listenerCount++
+	}
+
+	switch listenerCount {
+	case 0:
 		return errors.New("need at least one listener configured")
+	case 1:
+		// Only one listener is configured, do nothing
+		return nil
+	default:
+		// More than one listener is configured, return an error
+		return fmt.Errorf("only one listener can be configured at a time, found %d", listenerCount)
 	}
+}
 
-	if c.Processor.Kafka == nil && c.Processor.Search == nil && c.Processor.Webhook == nil && c.Processor.Postgres == nil {
-		return errors.New("need at least one processor configured")
+func (c *ProcessorConfig) IsValid() error {
+	processorCount := 0
+	if c.Kafka != nil {
+		processorCount++
+	}
+	if c.Postgres != nil {
+		processorCount++
+	}
+	if c.Search != nil {
+		processorCount++
+	}
+	if c.Webhook != nil {
+		processorCount++
 	}
 
-	return nil
+	switch processorCount {
+	case 0:
+		return errors.New("need at least one processor configured")
+	case 1:
+		// Only one processor is configured, do nothing
+		return nil
+	default:
+		// More than one processor is configured, return an error
+		return fmt.Errorf("only one processor can be configured at a time, found %d", processorCount)
+	}
 }
diff --git a/pkg/wal/processor/transformer/wal_transformer.go b/pkg/wal/processor/transformer/wal_transformer.go
index d611c74..9bbdbad 100644
--- a/pkg/wal/processor/transformer/wal_transformer.go
+++ b/pkg/wal/processor/transformer/wal_transformer.go
@@ -23,7 +23,7 @@ type Transformer struct {
 type columnTransformers map[string]transformers.Transformer
 
 type Config struct {
-	TransformerRulesFile string
+	TransformerRules []TableRules
 }
 
 type Option func(t *Transformer)
@@ -31,12 +31,7 @@ type Option func(t *Transformer)
 // New will return a transformer processor wrapper that will transform incoming
 // wal event column values as configured by the transformation rules.
 func New(cfg *Config, processor processor.Processor, opts ...Option) (*Transformer, error) {
-	rules, err := readRulesFromFile(cfg.TransformerRulesFile)
-	if err != nil {
-		return nil, err
-	}
-
-	transformerMap, err := transformerMapFromRules(rules)
+	transformerMap, err := transformerMapFromRules(cfg.TransformerRules)
 	if err != nil {
 		return nil, err
 	}
@@ -133,10 +128,10 @@ func schemaTableKey(schema, table string) string {
 	return schema + "/" + table
 }
 
-func transformerMapFromRules(rules *Rules) (map[string]columnTransformers, error) {
+func transformerMapFromRules(rules []TableRules) (map[string]columnTransformers, error) {
 	var err error
 	transformerMap := map[string]columnTransformers{}
-	for _, table := range rules.Transformers {
+	for _, table := range rules {
 		schemaTableTransformers := make(map[string]transformers.Transformer)
 		transformerMap[schemaTableKey(table.Schema, table.Table)] = schemaTableTransformers
 		for colName, transformerRules := range table.ColumnRules {
diff --git a/pkg/wal/processor/transformer/wal_transformer_rules.go b/pkg/wal/processor/transformer/wal_transformer_rules.go
index 7d5dd6c..29b01df 100644
--- a/pkg/wal/processor/transformer/wal_transformer_rules.go
+++ b/pkg/wal/processor/transformer/wal_transformer_rules.go
@@ -2,15 +2,6 @@
 
 package transformer
 
-import (
-	"fmt"
-	"os"
-
-	"gopkg.in/yaml.v3"
-)
-
-type RulesConfig struct{}
-
 type Rules struct {
 	Transformers []TableRules `yaml:"transformations"`
 }
@@ -26,17 +17,3 @@ type TransformerRules struct {
 	Parameters        map[string]any `yaml:"parameters"`
 	DynamicParameters map[string]any `yaml:"dynamic_parameters"`
 }
-
-func readRulesFromFile(filePath string) (*Rules, error) {
-	yamlFile, err := os.ReadFile(filePath)
-	if err != nil {
-		return nil, fmt.Errorf("reading rules from file: %w", err)
-	}
-
-	rules := &Rules{}
-	err = yaml.Unmarshal(yamlFile, &rules)
-	if err != nil {
-		return nil, fmt.Errorf("unmarshaling yaml file into transformer rules: %w", err)
-	}
-	return rules, nil
-}


Test patch:
diff --git a/cmd/config/config_env_test.go b/cmd/config/config_env_test.go
new file mode 100644
index 0000000..0a8bc79
--- /dev/null
+++ b/cmd/config/config_env_test.go
@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_EnvConfigToStreamConfig(t *testing.T) {
+	require.NoError(t, LoadFile("test/test_config.env"))
+
+	streamConfig, err := envConfigToStreamConfig()
+	assert.NoError(t, err)
+	assert.NotNil(t, streamConfig)
+
+	validateTestStreamConfig(t, streamConfig)
+}
diff --git a/cmd/config/config_yaml_test.go b/cmd/config/config_yaml_test.go
new file mode 100644
index 0000000..3aedd11
--- /dev/null
+++ b/cmd/config/config_yaml_test.go
@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
+)
+
+func TestYAMLConfig_toStreamConfig(t *testing.T) {
+	require.NoError(t, LoadFile("test/test_config.yaml"))
+
+	var config YAMLConfig
+	err := viper.Unmarshal(&config)
+	require.NoError(t, err)
+
+	streamConfig, err := config.toStreamConfig()
+	require.NoError(t, err)
+
+	validateTestStreamConfig(t, streamConfig)
+}
+
+func TestYAMLConfig_toStreamConfig_ErrorCases(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		config  YAMLConfig
+		wantErr error
+	}{
+		{
+			name: "err - invalid postgres source mode",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: "invalid",
+					},
+				},
+			},
+
+			wantErr: errUnsupportedPostgresSourceMode,
+		},
+		{
+			name: "err - invalid postgres snapshot mode",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: snapshotMode,
+						Snapshot: &SnapshotConfig{
+							Mode: "invalid",
+						},
+					},
+				},
+			},
+
+			wantErr: errUnsupportedSnapshotMode,
+		},
+		{
+			name: "err - invalid postgres snapshot schema mode",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: snapshotMode,
+						Snapshot: &SnapshotConfig{
+							Mode: schemaSnapshotMode,
+							Schema: &SnapshotSchemaConfig{
+								Mode: "invalid",
+							},
+						},
+					},
+				},
+			},
+
+			wantErr: errUnsupportedSchemaSnapshotMode,
+		},
+		{
+			name: "err - invalid postgres snapshot schema mode",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: snapshotMode,
+						Snapshot: &SnapshotConfig{
+							Mode: schemaSnapshotMode,
+							Schema: &SnapshotSchemaConfig{
+								Mode: "invalid",
+							},
+						},
+					},
+				},
+			},
+
+			wantErr: errUnsupportedSchemaSnapshotMode,
+		},
+		{
+			name: "err - invalid pgdump pgrestore config",
+			config: YAMLConfig{
+				Source: SourceConfig{
+					Postgres: &PostgresConfig{
+						Mode: snapshotMode,
+						Snapshot: &SnapshotConfig{
+							Mode: schemaSnapshotMode,
+							Schema: &SnapshotSchemaConfig{
+								Mode: pgdumprestoreSchemaMode,
+								PgDumpPgRestore: &PgDumpPgRestoreConfig{
+									CleanTargetDB: false,
+								},
+							},
+						},
+					},
+				},
+			},
+
+			wantErr: errInvalidPgdumpPgrestoreConfig,
+		},
+		{
+			name: "err - invalid injector config",
+			config: YAMLConfig{
+				Modifiers: ModifiersConfig{
+					Injector: &InjectorConfig{
+						Enabled: true,
+					},
+				},
+			},
+
+			wantErr: errInvalidInjectorConfig,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := tt.config.toStreamConfig()
+			require.ErrorIs(t, err, tt.wantErr)
+		})
+	}
+}
diff --git a/cmd/config/helper_test.go b/cmd/config/helper_test.go
new file mode 100644
index 0000000..11cc4c4
--- /dev/null
+++ b/cmd/config/helper_test.go
@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/xataio/pgstream/pkg/stream"
+)
+
+// this function validates the stream configuration produced from the test
+// configuration in the test directory.
+func validateTestStreamConfig(t *testing.T, streamConfig *stream.Config) {
+	assert.NotNil(t, streamConfig.Listener)
+	assert.NotNil(t, streamConfig.Listener.Postgres)
+	assert.Equal(t, "postgresql://user:password@localhost:5432/mydatabase", streamConfig.Listener.Postgres.Replication.PostgresURL)
+	assert.Equal(t, "pgstream_mydatabase_slot", streamConfig.Listener.Postgres.Replication.ReplicationSlotName)
+	assert.NotNil(t, streamConfig.Listener.Postgres.Snapshot)
+	assert.ElementsMatch(t, []string{"test", "test_schema.test", "another_schema.*"}, streamConfig.Listener.Postgres.Snapshot.Adapter.Tables)
+	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Adapter.SnapshotWorkers)
+	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Generator.SchemaWorkers)
+	assert.Equal(t, uint(4), streamConfig.Listener.Postgres.Snapshot.Generator.TableWorkers)
+	assert.Equal(t, uint(1000), streamConfig.Listener.Postgres.Snapshot.Generator.BatchPageSize)
+	assert.NotNil(t, streamConfig.Listener.Postgres.Snapshot.Schema.DumpRestore)
+	assert.True(t, streamConfig.Listener.Postgres.Snapshot.Schema.DumpRestore.CleanTargetDB)
+
+	assert.NotNil(t, streamConfig.Listener.Kafka)
+	assert.Equal(t, []string{"localhost:9092"}, streamConfig.Listener.Kafka.Reader.Conn.Servers)
+	assert.Equal(t, "mytopic", streamConfig.Listener.Kafka.Reader.Conn.Topic.Name)
+	assert.Equal(t, "mygroup", streamConfig.Listener.Kafka.Reader.ConsumerGroupID)
+	assert.Equal(t, "earliest", streamConfig.Listener.Kafka.Reader.ConsumerGroupStartOffset)
+	assert.True(t, streamConfig.Listener.Kafka.Reader.Conn.TLS.Enabled)
+	assert.Equal(t, "/path/to/ca.crt", streamConfig.Listener.Kafka.Reader.Conn.TLS.CaCertFile)
+	assert.Equal(t, "/path/to/client.crt", streamConfig.Listener.Kafka.Reader.Conn.TLS.ClientCertFile)
+	assert.Equal(t, "/path/to/client.key", streamConfig.Listener.Kafka.Reader.Conn.TLS.ClientKeyFile)
+	assert.Equal(t, uint(5), streamConfig.Listener.Kafka.Checkpointer.CommitBackoff.Exponential.MaxRetries)
+	assert.Equal(t, time.Second, streamConfig.Listener.Kafka.Checkpointer.CommitBackoff.Exponential.InitialInterval)
+	assert.Equal(t, 60*time.Second, streamConfig.Listener.Kafka.Checkpointer.CommitBackoff.Exponential.MaxInterval)
+
+	assert.NotNil(t, streamConfig.Processor.Postgres)
+	assert.Equal(t, "postgresql://user:password@localhost:5432/mytargetdatabase", streamConfig.Processor.Postgres.BatchWriter.URL)
+	assert.Equal(t, int64(100), streamConfig.Processor.Postgres.BatchWriter.BatchConfig.MaxBatchSize)
+	assert.Equal(t, time.Second, streamConfig.Processor.Postgres.BatchWriter.BatchConfig.BatchTimeout)
+	assert.Equal(t, int64(1572864), streamConfig.Processor.Postgres.BatchWriter.BatchConfig.MaxBatchBytes)
+	assert.Equal(t, int64(204800), streamConfig.Processor.Postgres.BatchWriter.BatchConfig.MaxQueueBytes)
+	assert.False(t, streamConfig.Processor.Postgres.BatchWriter.DisableTriggers)
+	assert.Equal(t, "nothing", streamConfig.Processor.Postgres.BatchWriter.OnConflictAction)
+
+	assert.NotNil(t, streamConfig.Processor.Kafka)
+	assert.Equal(t, "mytopic", streamConfig.Processor.Kafka.Writer.Kafka.Topic.Name)
+	assert.Equal(t, 1, streamConfig.Processor.Kafka.Writer.Kafka.Topic.NumPartitions)
+	assert.Equal(t, 1, streamConfig.Processor.Kafka.Writer.Kafka.Topic.ReplicationFactor)
+	assert.True(t, streamConfig.Processor.Kafka.Writer.Kafka.Topic.AutoCreate)
+	assert.Equal(t, []string{"localhost:9092"}, streamConfig.Processor.Kafka.Writer.Kafka.Servers)
+	assert.Equal(t, int64(100), streamConfig.Processor.Kafka.Writer.Batch.MaxBatchSize)
+	assert.Equal(t, time.Second, streamConfig.Processor.Kafka.Writer.Batch.BatchTimeout)
+	assert.Equal(t, int64(1572864), streamConfig.Processor.Kafka.Writer.Batch.MaxBatchBytes)
+	assert.Equal(t, int64(204800), streamConfig.Processor.Kafka.Writer.Batch.MaxQueueBytes)
+	assert.True(t, streamConfig.Processor.Kafka.Writer.Kafka.TLS.Enabled)
+	assert.Equal(t, "/path/to/ca.crt", streamConfig.Processor.Kafka.Writer.Kafka.TLS.CaCertFile)
+	assert.Equal(t, "/path/to/client.crt", streamConfig.Processor.Kafka.Writer.Kafka.TLS.ClientCertFile)
+	assert.Equal(t, "/path/to/client.key", streamConfig.Processor.Kafka.Writer.Kafka.TLS.ClientKeyFile)
+
+	assert.NotNil(t, streamConfig.Processor.Search)
+	assert.Equal(t, "http://localhost:9200", streamConfig.Processor.Search.Store.ElasticsearchURL)
+	assert.Equal(t, int64(100), streamConfig.Processor.Search.Indexer.Batch.MaxBatchSize)
+	assert.Equal(t, time.Second, streamConfig.Processor.Search.Indexer.Batch.BatchTimeout)
+	assert.Equal(t, int64(1572864), streamConfig.Processor.Search.Indexer.Batch.MaxBatchBytes)
+	assert.Equal(t, int64(204800), streamConfig.Processor.Search.Indexer.Batch.MaxQueueBytes)
+	assert.Equal(t, uint(5), streamConfig.Processor.Search.Retrier.Backoff.Exponential.MaxRetries)
+	assert.Equal(t, time.Second, streamConfig.Processor.Search.Retrier.Backoff.Exponential.InitialInterval)
+	assert.Equal(t, 60*time.Second, streamConfig.Processor.Search.Retrier.Backoff.Exponential.MaxInterval)
+
+	assert.NotNil(t, streamConfig.Processor.Webhook)
+	assert.Equal(t, "localhost:9090", streamConfig.Processor.Webhook.SubscriptionServer.Address)
+	assert.Equal(t, uint(4), streamConfig.Processor.Webhook.Notifier.URLWorkerCount)
+	assert.Equal(t, "postgresql://user:password@localhost:5432/mydatabase", streamConfig.Processor.Webhook.SubscriptionStore.URL)
+	assert.True(t, streamConfig.Processor.Webhook.SubscriptionStore.CacheEnabled)
+	assert.Equal(t, 60*time.Second, streamConfig.Processor.Webhook.SubscriptionStore.CacheRefreshInterval)
+	assert.Equal(t, time.Minute, streamConfig.Processor.Webhook.SubscriptionServer.ReadTimeout)
+	assert.Equal(t, time.Minute, streamConfig.Processor.Webhook.SubscriptionServer.WriteTimeout)
+	assert.Equal(t, time.Second, streamConfig.Processor.Webhook.Notifier.ClientTimeout)
+
+	assert.NotNil(t, streamConfig.Processor.Injector)
+	assert.Equal(t, "postgresql://user:password@localhost:5432/mydatabase", streamConfig.Processor.Injector.Store.URL)
+
+	assert.NotNil(t, streamConfig.Processor.Transformer)
+	assert.Len(t, streamConfig.Processor.Transformer.TransformerRules, 1)
+	assert.Equal(t, "public", streamConfig.Processor.Transformer.TransformerRules[0].Schema)
+	assert.Equal(t, "test", streamConfig.Processor.Transformer.TransformerRules[0].Table)
+	assert.NotNil(t, streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules)
+	assert.NotNil(t, streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules["name"])
+	assert.Equal(t, "greenmask_firstname", streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules["name"].Name)
+	assert.NotNil(t, streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules["name"].DynamicParameters)
+	assert.NotNil(t, streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules["name"].DynamicParameters["gender"])
+	dynParamGender, ok := streamConfig.Processor.Transformer.TransformerRules[0].ColumnRules["name"].DynamicParameters["gender"].(map[string]any)
+	assert.True(t, ok)
+	column, ok := dynParamGender["column"].(string)
+	assert.True(t, ok)
+	assert.Equal(t, "sex", column)
+}
diff --git a/cmd/config/test/test_config.env b/cmd/config/test/test_config.env
new file mode 100644
index 0000000..e1197e2
--- /dev/null
+++ b/cmd/config/test/test_config.env
@@ -0,0 +1,83 @@
+#### Source ####
+
+# Postgres
+PGSTREAM_POSTGRES_LISTENER_URL="postgresql://user:password@localhost:5432/mydatabase"
+PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME="pgstream_mydatabase_slot"
+PGSTREAM_POSTGRES_LISTENER_INITIAL_SNAPSHOT_ENABLED=true
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_STORE_URL="postgresql://user:password@localhost:5432/mydatabase"
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLES="test test_schema.test another_schema.*"
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_WORKERS=4
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_SCHEMA_WORKERS=4
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_TABLE_WORKERS=4
+PGSTREAM_POSTGRES_INITIAL_SNAPSHOT_BATCH_PAGE_SIZE=1000
+PGSTREAM_POSTGRES_SNAPSHOT_CLEAN_TARGET_DB=true
+
+# Kafka
+PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_TOPIC_NAME="mytopic"
+PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID="mygroup"
+PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET="earliest"
+PGSTREAM_KAFKA_COMMIT_EXP_BACKOFF_INITIAL_INTERVAL="1s"
+PGSTREAM_KAFKA_COMMIT_EXP_BACKOFF_MAX_INTERVAL="1m"
+PGSTREAM_KAFKA_COMMIT_EXP_BACKOFF_MAX_RETRIES=5
+PGSTREAM_KAFKA_TLS_ENABLED=true
+PGSTREAM_KAFKA_TLS_CA_CERT_FILE="/path/to/ca.crt"
+PGSTREAM_KAFKA_TLS_CLIENT_CERT_FILE="/path/to/client.crt"
+PGSTREAM_KAFKA_TLS_CLIENT_KEY_FILE="/path/to/client.key"
+
+
+#### Target ####
+
+# Postgres
+PGSTREAM_POSTGRES_WRITER_TARGET_URL="postgresql://user:password@localhost:5432/mytargetdatabase"
+PGSTREAM_POSTGRES_WRITER_BATCH_SIZE=100
+PGSTREAM_POSTGRES_WRITER_BATCH_TIMEOUT="1s"
+PGSTREAM_POSTGRES_WRITER_MAX_QUEUE_BYTES=204800
+PGSTREAM_POSTGRES_WRITER_BATCH_BYTES=1572864
+PGSTREAM_POSTGRES_WRITER_SCHEMALOG_STORE_URL="postgresql://user:password@localhost:5432/mydatabase"
+PGSTREAM_POSTGRES_WRITER_DISABLE_TRIGGERS=false
+PGSTREAM_POSTGRES_WRITER_ON_CONFLICT_ACTION="nothing"
+
+# Kafka
+PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_TOPIC_NAME="mytopic"
+PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
+PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1
+PGSTREAM_KAFKA_TOPIC_AUTO_CREATE=true
+PGSTREAM_KAFKA_WRITER_BATCH_TIMEOUT="1s"
+PGSTREAM_KAFKA_WRITER_BATCH_SIZE=100
+PGSTREAM_KAFKA_WRITER_BATCH_BYTES=1572864
+PGSTREAM_KAFKA_WRITER_MAX_QUEUE_BYTES=204800
+PGSTREAM_KAFKA_TLS_ENABLED=true
+PGSTREAM_KAFKA_TLS_CA_CERT_FILE="/path/to/ca.crt"
+PGSTREAM_KAFKA_TLS_CLIENT_CERT_FILE="/path/to/client.crt"
+PGSTREAM_KAFKA_TLS_CLIENT_KEY_FILE="/path/to/client.key"
+
+
+# Search
+PGSTREAM_SEARCH_INDEXER_BATCH_SIZE=100
+PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT="1s"
+PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES=204800
+PGSTREAM_SEARCH_INDEXER_BATCH_BYTES=1572864
+PGSTREAM_ELASTICSEARCH_STORE_URL="http://localhost:9200"
+PGSTREAM_SEARCH_STORE_EXP_BACKOFF_INITIAL_INTERVAL="1s"
+PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_INTERVAL="1m"
+PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_RETRIES=5
+
+# Webhooks
+PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_URL="postgresql://user:password@localhost:5432/mydatabase"
+PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_ENABLED=true
+PGSTREAM_WEBHOOK_SUBSCRIPTION_STORE_CACHE_REFRESH_INTERVAL="60s"
+PGSTREAM_WEBHOOK_SUBSCRIPTION_SERVER_ADDRESS="localhost:9090"
+PGSTREAM_WEBHOOK_SUBSCRIPTION_SERVER_READ_TIMEOUT="60s"
+PGSTREAM_WEBHOOK_SUBSCRIPTION_SERVER_WRITE_TIMEOUT="60s"
+PGSTREAM_WEBHOOK_NOTIFIER_WORKER_COUNT=4
+PGSTREAM_WEBHOOK_NOTIFIER_CLIENT_TIMEOUT="1s"
+
+#### Modifiers ####
+
+# Injector
+PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgresql://user:password@localhost:5432/mydatabase"
+
+# Transformers
+PGSTREAM_TRANSFORMER_RULES_FILE="test/test_transformer_rules.yaml"
diff --git a/cmd/config/test/test_config.yaml b/cmd/config/test/test_config.yaml
new file mode 100644
index 0000000..4625bfb
--- /dev/null
+++ b/cmd/config/test/test_config.yaml
@@ -0,0 +1,109 @@
+source:
+  postgres:
+    url: "postgresql://user:password@localhost:5432/mydatabase"
+    mode: snapshot_and_replication # options are replication, snapshot or snapshot_and_replication
+    snapshot: # when mode is snapshot or snapshot_and_replication
+      mode: full # options are data_and, schema or data
+      tables: ["test", "test_schema.test", "another_schema.*"] # tables to snapshot, can be a list of table names or a pattern
+      recorder_enabled: true # whether to record snapshot status in a table under the pgstream schema in the source database
+      snapshot_workers: 4 # number of schemas to be snapshotted in parallel
+      data: # when mode is full or data
+        schema_workers: 4 # number of schema tables to be snapshotted in parallel
+        table_workers: 4 # number of workers to snapshot a table in parallel
+        batch_page_size: 1000 # number of pages to read per batch
+      schema: # when mode is full or schema
+        mode: pgdump_pgrestore # options are pgdump_pgrestore or schemalog
+        pgdump_pgrestore:
+          clean_target_db: true # whether to clean the target database before restoring
+    replication: # when mode is replication or snapshot_and_replication
+      replication_slot: "pgstream_mydatabase_slot"
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "mytopic"
+    consumer_group:
+      id: "mygroup"
+      start_offset: "earliest" # options are earliest or latest
+    tls:
+      ca_cert: "/path/to/ca.crt" # path to CA certificate
+      client_cert: "/path/to/client.crt" # path to client certificate
+      client_key: "/path/to/client.key" # path to client key
+    backoff:
+      exponential:
+        max_retries: 5 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+      constant:
+        max_retries: 5 # maximum number of retries
+        interval: 1000 # interval in milliseconds
+
+target:
+  postgres:
+    url: "postgresql://user:password@localhost:5432/mytargetdatabase"
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+    disable_triggers: false # whether to disable triggers on the target database
+    on_conflict_action: "nothing" # options are update, nothing or error
+  kafka:
+    servers: ["localhost:9092"]
+    topic:
+      name: "mytopic" # name of the Kafka topic
+      partitions: 1 # number of partitions for the topic
+      replication_factor: 1 # replication factor for the topic
+      auto_create: true # whether to automatically create the topic if it doesn't exist
+    tls:
+      ca_cert: "/path/to/ca.crt" # path to CA certificate
+      client_cert: "/path/to/client.crt" # path to client certificate
+      client_key: "/path/to/client.key" # path to client key
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+  search:
+    engine: "elasticsearch" # options are elasticsearch or opensearch
+    url: "http://localhost:9200" # URL of the search engine
+    batch:
+      timeout: 1000 # batch timeout in milliseconds
+      size: 100 # number of messages in a batch
+      max_bytes: 1572864 # max size of batch in bytes (1.5MiB)
+      max_queue_bytes: 204800 # max size of memory guard queue in bytes (100MiB)
+    backoff:
+      exponential:
+        max_retries: 5 # maximum number of retries
+        initial_interval: 1000 # initial interval in milliseconds
+        max_interval: 60000 # maximum interval in milliseconds
+      constant:
+        max_retries: 5 # maximum number of retries
+        interval: 1000 # interval in milliseconds
+  webhooks:
+    subscriptions:
+      store:
+        url: "postgresql://user:password@localhost:5432/mydatabase" # URL of the database where the webhook subscriptions are stored
+        cache:
+          enabled: true # whether to enable caching for the subscription store
+          refresh_interval: 60 # interval in seconds to refresh the cache
+      server:
+        address: "localhost:9090" # address of the subscription server
+        read_timeout: 60 # read timeout in seconds
+        write_timeout: 60 # write timeout in seconds
+    notifier:
+      worker_count: 4 # number of notifications to be processed in parallel
+      client_timeout: 1000 # timeout for the webhook client in milliseconds
+
+
+modifiers:
+  injector:
+    enabled: true # whether to inject pgstream metadata into the WAL events
+  transformations:
+  - schema: public
+    table: test
+    column_transformers:
+      name:
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex
diff --git a/cmd/config/test/test_transformer_rules.yaml b/cmd/config/test/test_transformer_rules.yaml
new file mode 100644
index 0000000..891cbe7
--- /dev/null
+++ b/cmd/config/test/test_transformer_rules.yaml
@@ -0,0 +1,9 @@
+transformations:
+  - schema: public
+    table: test
+    column_transformers:
+      name:
+        name: greenmask_firstname
+        dynamic_parameters:
+          gender:
+            column: sex
diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go
index e3db40a..0196e9f 100644
--- a/pkg/stream/integration/helper_test.go
+++ b/pkg/stream/integration/helper_test.go
@@ -262,7 +262,7 @@ func testPostgresProcessorCfgWithTransformer(sourcePGURL string) stream.Processo
 			},
 		},
 		Transformer: &transformer.Config{
-			TransformerRulesFile: "config/integration_test_transformer_rules.yaml",
+			TransformerRules: testTransformationRules(),
 		},
 	}
 }
@@ -279,3 +279,98 @@ func testKafkaCfg() kafkalib.ConnConfig {
 		},
 	}
 }
+
+func testTransformationRules() []transformer.TableRules {
+	return []transformer.TableRules{
+		{
+			Schema: "public",
+			Table:  "pg2pg_integration_transformer_test",
+			ColumnRules: map[string]transformer.TransformerRules{
+				"name": {
+					Name: "neosync_firstname",
+					Parameters: map[string]any{
+						"preserve_length": false,
+						"max_length":      5,
+					},
+				},
+				"last_name": {
+					Name: "neosync_string",
+					Parameters: map[string]any{
+						"preserve_length": false,
+						"max_length":      10,
+					},
+				},
+				"email": {
+					Name: "neosync_email",
+					Parameters: map[string]any{
+						"preserve_length":  false,
+						"max_length":       15,
+						"excluded_domains": []string{"example.com", "example.net"},
+					},
+				},
+				"secondary_email": {
+					Name: "masking",
+					Parameters: map[string]any{
+						"type": "email",
+					},
+				},
+				"address": {
+					Name: "greenmask_string",
+					Parameters: map[string]any{
+						"preserve_length": false,
+						"max_length":      20,
+					},
+				},
+				"age": {
+					Name: "greenmask_integer",
+					Parameters: map[string]any{
+						"generator": "deterministic",
+						"min_value": 18,
+						"max_value": 75,
+					},
+				},
+				"total_purchases": {
+					Name: "greenmask_float",
+					Parameters: map[string]any{
+						"generator": "deterministic",
+						"min_value": 0.0,
+						"max_value": 1000.0,
+					},
+				},
+				"customer_id": {
+					Name: "greenmask_uuid",
+				},
+				"birth_date": {
+					Name: "greenmask_date",
+					Parameters: map[string]any{
+						"min_value": "1990-01-01",
+						"max_value": "2000-12-31",
+					},
+				},
+				"is_active": {
+					Name: "greenmask_boolean",
+				},
+				"created_at": {
+					Name: "greenmask_unix_timestamp",
+					Parameters: map[string]any{
+						"min_value": "1741856058",
+						"max_value": "1741956058",
+					},
+				},
+				"updated_at": {
+					Name: "greenmask_utc_timestamp",
+					Parameters: map[string]any{
+						"min_timestamp": "2022-01-01T00:00:00Z",
+						"max_timestamp": "2024-01-01T23:59:59Z",
+					},
+				},
+				"gender": {
+					Name: "greenmask_choice",
+					Parameters: map[string]any{
+						"choices": []string{"M", "F", "None"},
+					},
+				},
+			},
+		},
+	}
+}
diff --git a/pkg/wal/processor/transformer/test/test_transformer_rules.yaml b/pkg/wal/processor/transformer/test/test_transformer_rules.yaml
deleted file mode 100644
index b25d700..0000000
--- a/pkg/wal/processor/transformer/test/test_transformer_rules.yaml
+++ /dev/null
@@ -1,18 +0,0 @@
-transformations:
-  - schema: public
-    table: test1
-    column_transformers:
-      column_1:
-        name: string
-        generator: random
-        parameters:
-          min_length: 1
-          max_length: 2
-  - schema: test
-    table: test2
-    column_transformers:
-      column_2:
-        name: string
-        generator: deterministic
-        parameters:
-          symbols: "abcdef"
diff --git a/pkg/wal/processor/transformer/wal_transformer_rules_test.go b/pkg/wal/processor/transformer/wal_transformer_rules_test.go
deleted file mode 100644
index bcf73f7..0000000
--- a/pkg/wal/processor/transformer/wal_transformer_rules_test.go
+++ /dev/null
@@ -1,66 +0,0 @@
-// SPDX-License-Identifier: Apache-2.0
-
-package transformer
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/require"
-)
-
-func Test_readRulesFromFile(t *testing.T) {
-	t.Parallel()
-
-	tests := []struct {
-		name     string
-		filename string
-
-		wantRules *Rules
-		wantErr   error
-	}{
-		{
-			name:     "ok",
-			filename: "test/test_transformer_rules.yaml",
-			wantRules: &Rules{
-				Transformers: []TableRules{
-					{
-						Schema: "public",
-						Table:  "test1",
-						ColumnRules: map[string]TransformerRules{
-							"column_1": {
-								Name: "string",
-								Parameters: map[string]any{
-									"min_length": 1,
-									"max_length": 2,
-								},
-							},
-						},
-					},
-					{
-						Schema: "test",
-						Table:  "test2",
-						ColumnRules: map[string]TransformerRules{
-							"column_2": {
-								Name: "string",
-								Parameters: map[string]any{
-									"symbols": "abcdef",
-								},
-							},
-						},
-					},
-				},
-			},
-			wantErr: nil,
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			rules, err := readRulesFromFile(tc.filename)
-			require.ErrorIs(t, err, tc.wantErr)
-			require.Equal(t, tc.wantRules, rules)
-		})
-	}
-}
diff --git a/pkg/wal/processor/transformer/wal_transformer_test.go b/pkg/wal/processor/transformer/wal_transformer_test.go
index e0f0396..a5d7bca 100644
--- a/pkg/wal/processor/transformer/wal_transformer_test.go
+++ b/pkg/wal/processor/transformer/wal_transformer_test.go
@@ -33,7 +33,26 @@ func TestTransformer_New(t *testing.T) {
 		{
 			name: "ok",
 			config: &Config{
-				TransformerRulesFile: "test/test_transformer_rules.yaml",
+				TransformerRules: []TableRules{
+					{
+						Schema: "public",
+						Table:  "test1",
+						ColumnRules: map[string]TransformerRules{
+							"column_1": {
+								Name: "string",
+							},
+						},
+					},
+					{
+						Schema: "test",
+						Table:  "test2",
+						ColumnRules: map[string]TransformerRules{
+							"column_2": {
+								Name: "string",
+							},
+						},
+					},
+				},
 			},
 
 			wantTransformer: &Transformer{
@@ -235,25 +254,23 @@ func Test_transformerMapFromRules(t *testing.T) {
 
 	tests := []struct {
 		name  string
-		rules *Rules
+		rules []TableRules
 
 		wantTransformerMap map[string]columnTransformers
 		wantErr            error
 	}{
 		{
 			name: "ok",
-			rules: &Rules{
-				Transformers: []TableRules{
-					{
-						Schema: testSchema,
-						Table:  testTable,
-						ColumnRules: map[string]TransformerRules{
-							"column_1": {
-								Name: "string",
-							},
-							"column_2": {
-								Name: "string",
-							},
+			rules: []TableRules{
+				{
+					Schema: testSchema,
+					Table:  testTable,
+					ColumnRules: map[string]TransformerRules{
+						"column_1": {
+							Name: "string",
+						},
+						"column_2": {
+							Name: "string",
 						},
 					},
 				},
@@ -269,22 +286,20 @@ func Test_transformerMapFromRules(t *testing.T) {
 		},
 		{
 			name:  "ok - no rules",
-			rules: &Rules{},
+			rules: []TableRules{},
 
 			wantTransformerMap: map[string]columnTransformers{},
 			wantErr:            nil,
 		},
 		{
 			name: "error - invalid transformer rules",
-			rules: &Rules{
-				Transformers: []TableRules{
-					{
-						Schema: testSchema,
-						Table:  testTable,
-						ColumnRules: map[string]TransformerRules{
-							"column_1": {
-								Name: "invalid",
-							},
+			rules: []TableRules{
+				{
+					Schema: testSchema,
+					Table:  testTable,
+					ColumnRules: map[string]TransformerRules{
+						"column_1": {
+							Name: "invalid",
 						},
 					},
 				},