use strict;
use warnings FATAL => 'all';

use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;

my $tag = $$;
my $primary = PostgreSQL::Test::Cluster->new("repro_ip_primary_$tag");

$primary->init(allows_streaming => 'logical');
$primary->append_conf(
	'postgresql.conf', qq(
autovacuum = off
log_min_messages = 'debug2'
));
$primary->start;

if (!$primary->check_extension('injection_points'))
{
	plan skip_all => 'Extension injection_points not installed';
}

# Create the extension before the base backup so the standby can call its
# functions while in recovery.
$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points;');

my $backup_name = 'backup';
$primary->backup($backup_name);

my $standby = PostgreSQL::Test::Cluster->new("repro_ip_standby_$tag");
$standby->init_from_backup(
	$primary, $backup_name,
	has_streaming => 1,
	has_restoring => 1);

my $primary_connstr = $primary->connstr;
$standby->append_conf(
	'postgresql.conf', qq(
hot_standby_feedback = on
primary_slot_name = 'phys_slot'
primary_conninfo = '$primary_connstr dbname=postgres'
log_min_messages = 'debug2'
));

$primary->safe_psql(
	'postgres',
	q{
SELECT pg_create_logical_replication_slot('victim_slot', 'pgoutput', false, false, true);
SELECT pg_create_physical_replication_slot('phys_slot');
});

$standby->start;
$primary->advance_wal(1);
$primary->wait_for_replay_catchup($standby);

note('attach injection point on standby');
$standby->safe_psql(
	'postgres',
	q{SELECT injection_points_attach('slotsync-obsolete-after-drop', 'wait');}
);

note('sync failover slot to standby');
$standby->safe_psql('postgres', 'SELECT pg_sync_replication_slots();');

is( $standby->safe_psql(
		'postgres',
		q{SELECT synced FROM pg_replication_slots WHERE slot_name = 'victim_slot';}
	),
	't',
	'victim slot is synchronized to the standby');

note('drop remote failover slot');
$primary->safe_psql('postgres',
	q{SELECT pg_drop_replication_slot('victim_slot');});

my $log_offset = -s $standby->logfile;

note('start slot sync and wait at injection point after local drop');
my $sync = $standby->background_psql('postgres', on_error_stop => 0);
$sync->query_until(
	qr/start_sync/,
	q(
\echo start_sync
SELECT pg_sync_replication_slots();
));

$standby->wait_for_event('client backend', 'slotsync-obsolete-after-drop');

ok( $standby->poll_query_until(
		'postgres',
		q{SELECT NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'victim_slot');}
	),
	'victim slot has been dropped locally');

note('reuse freed slot array cell with a physical slot');
$standby->safe_psql('postgres',
	q{SELECT pg_create_physical_replication_slot('replacement_slot');});

note('release injection point');
$standby->safe_psql(
	'postgres',
	q{
SELECT injection_points_wakeup('slotsync-obsolete-after-drop');
SELECT injection_points_detach('slotsync-obsolete-after-drop');
});

ok( $standby->wait_for_log(
		qr/dropped replication slot "replacement_slot" of database with OID 0|you don't own a lock of type AccessShareLock/,
		$log_offset),
	'stale local_slot pointer was observed after drop');

is( $standby->safe_psql(
		'postgres',
		q{SELECT slot_type FROM pg_replication_slots WHERE slot_name = 'replacement_slot';}
	),
	'physical',
	'replacement slot still exists');

$sync->quit;

done_testing();