use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;

program_help_ok('pg_createsubscriber');
program_version_ok('pg_createsubscriber');
program_options_handling_ok('pg_createsubscriber');

# Primary
my $primary = PostgreSQL::Test::Cluster->new('primary');
my $primary_connstr = $primary->connstr;
$primary->init(allows_streaming => 'logical');
$primary->start();

# Setup test database and physical replication slot
$primary->safe_psql('postgres', "CREATE DATABASE testdb");
$primary->safe_psql('testdb',
    "CREATE TABLE test_data (id serial PRIMARY KEY, value text);");
$primary->safe_psql('testdb',
    "INSERT INTO test_data (value) VALUES ('initial_data');");
$primary->safe_psql('testdb',
    "SELECT pg_create_physical_replication_slot('physical_slot');");

# Create backup for physical replica
$primary->backup('primary_backup');

# Physical replica from backup
my $physical_replica = PostgreSQL::Test::Cluster->new('physical_replica');
my $physical_replica_connstr = $physical_replica->connstr;
$physical_replica->init_from_backup($primary, 'primary_backup', has_streaming => 1);

# Configure physical replication
$physical_replica->append_conf('postgresql.conf', qq[
primary_slot_name = 'physical_slot'
primary_conninfo = '$primary_connstr dbname=testdb'
hot_standby_feedback = on
]);

$physical_replica->set_standby_mode();
$physical_replica->start();

# Verify physical replication is working
my $result = $physical_replica->safe_psql('testdb',
    "SELECT count(*) FROM test_data;");
is($result, '1', 'physical replication working - data replicated');
$physical_replica->stop();

# Convert physical replica to logical subscriber using pg_createsubscriber
command_ok(
    [
        'pg_createsubscriber', '--verbose',
        '-D', $physical_replica->data_dir,
        '--database', 'testdb',
        '--publisher-server', $primary_connstr,
        '--replication-slot', 'logical_slot',
        '--publication', 'test_publication',
        '--subscription', 'test_subscription'
    ],
    'convert physical replica to logical subscriber'
);

# Start the logical subscriber
$physical_replica->start();

# Create a physical standby from the promoted subscriber
$physical_replica->safe_psql('testdb',
    "SELECT pg_create_physical_replication_slot('new_physical_slot');");

# Create backup from promoted subscriber
$physical_replica->backup('promoted_primary_backup');

# Initialize new physical standby
my $new_standby = PostgreSQL::Test::Cluster->new('new_standby');
$new_standby->init_from_backup($physical_replica, 'promoted_primary_backup',
    has_streaming => 1);

# Configure the new standby
$new_standby->append_conf('postgresql.conf', qq[
primary_slot_name = 'new_physical_slot'
primary_conninfo = '$physical_replica_connstr dbname=testdb'
hot_standby_feedback = on
]);

$new_standby->set_standby_mode();
is($new_standby->start(), 1, "new replica performs recovery");

done_testing();