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

if ($ENV{enable_injection_points} ne 'yes')
{
	plan skip_all => 'Injection points not supported by this build';
}

my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;

my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;

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

$node_publisher->safe_psql('postgres', 'CREATE EXTENSION injection_points');

for my $node ($node_publisher, $node_subscriber)
{
	$node->safe_psql('postgres',
		'CREATE TABLE tab_nonri (id int PRIMARY KEY, val int)');
}

$node_publisher->safe_psql(
	'postgres', qq{
	CREATE PUBLICATION pub_sync FOR TABLE tab_nonri;
	CREATE PUBLICATION pub_filtered;
});

my $connstr = $node_publisher->connstr . ' dbname=postgres';

$node_subscriber->safe_psql(
	'postgres', "
	CREATE SUBSCRIPTION sub
	CONNECTION '$connstr application_name=sub'
	PUBLICATION pub_sync, pub_filtered");

$node_subscriber->wait_for_subscription_sync($node_publisher, 'sub');

# The subscriber must hold the row, so that failing to remove it is visible.
$node_publisher->safe_psql('postgres', 'INSERT INTO tab_nonri VALUES (1, 1)');
$node_publisher->wait_for_catchup('sub');

is( $node_subscriber->safe_psql(
		'postgres', 'SELECT val FROM tab_nonri WHERE id = 1'),
	'1',
	'the subscriber holds the row before the race');

# From here the table belongs to no publication, so nothing is published until
# the filtered publication picks it up during the race below.
$node_publisher->safe_psql('postgres',
	'ALTER PUBLICATION pub_sync DROP TABLE tab_nonri');

###############################################################################
# Hold the UPDATE after CheckCmdReplicaIdentity() has allowed it and before its
# WAL record is written, and commit a filter on a non-replica-identity column in
# that window.
###############################################################################

my $upd = $node_publisher->background_psql('postgres');

$upd->query_safe('SELECT injection_points_set_local()');
$upd->query_safe(
	"SELECT injection_points_attach('heap_update-before-pin', 'wait')");

# val leaves the filter's set; id, the replica identity, is untouched, so no old
# tuple is logged.  Issued without waiting, since it is about to park.
$upd->query_until(
	qr/^issued$/m, qq{
	\\echo issued
	UPDATE tab_nonri SET val = 2 WHERE id = 1;
});

$node_publisher->wait_for_event('client backend', 'heap_update-before-pin');

# val is not part of the replica identity, so this filter is one that
# CheckCmdReplicaIdentity() exists to forbid -- but the statement above has
# already passed that check.
$node_publisher->safe_psql('postgres',
	'ALTER PUBLICATION pub_filtered ADD TABLE tab_nonri WHERE (val = 1)');

is( $node_publisher->safe_psql(
		'postgres', q{
		SELECT count(*) FROM pg_publication_rel r
		  JOIN pg_publication p ON p.oid = r.prpubid
		 WHERE p.pubname = 'pub_filtered' AND r.prqual IS NOT NULL}),
	'1',
	'the row filter is committed while the update is parked');

$node_publisher->safe_psql(
	'postgres', "
	SELECT injection_points_wakeup('heap_update-before-pin');
	SELECT injection_points_detach('heap_update-before-pin');");

ok($upd->quit, 'the update completes');

$node_publisher->wait_for_catchup('sub');

###############################################################################
# Confirm the window really did let through a statement that is otherwise
# refused: any further UPDATE on this table is now rejected.
###############################################################################

my ($ret, $stdout, $stderr) =
  $node_publisher->psql('postgres',
	'UPDATE tab_nonri SET val = 3 WHERE id = 1');

isnt($ret, 0, 'a later UPDATE is refused, as the check intends');
like(
	$stderr,
	qr/Column used in the publication WHERE expression is not part of the replica identity/,
	'and refused for exactly the reason the parked UPDATE escaped');

###############################################################################
# The row left the filter's set, so the subscriber should have received a DELETE.
###############################################################################

is( $node_publisher->safe_psql(
		'postgres', 'SELECT val FROM tab_nonri WHERE id = 1'),
	'2',
	'the publisher applied the update');

is( $node_subscriber->safe_psql('postgres', 'SELECT count(*) FROM tab_nonri'),
	'0',
	'the row that left the filter is removed from the subscriber')
  or diag(
	'no old tuple was logged, because the replica identity did not change, so '
	  . 'pgoutput evaluated the filter against the new tuple alone, found it '
	  . 'did not match, and published nothing; the subscriber still holds the '
	  . 'pre-update row');

$node_subscriber->stop;
$node_publisher->stop;

done_testing();