Posgres CDC Trigger

cdc_trigger_test.sql

drop table  if exists transaction;
create table transaction (
  id SERIAL primary key,
  issued_at TIMESTAMP not null
);

drop table  if exists  employee;
create table employee (
id SERIAL primary key,
name varchar,
salary decimal
);

drop table  if exists employee_version;
create table employee_version (
id INT,
name varchar,
salary decimal,
transaction_id INT,
end_transaction_id INT,
operation_type INT,
 PRIMARY KEY(id, transaction_id)
);

create or replace function log_employee_changes ()
returns trigger as $$
	declare tx_id INT;
begin
	begin
		insert into transaction (issued_at) values (NOW()) returning id into tx_id;
	end;

	if (TG_OP='DELETE') then
		update employee_version set end_transaction_id = tx_id 
		where id in (select id from OLD_TABLE);

		insert into employee_version (id, name, salary, transaction_id, operation_type)
		select id, name, salary, tx_id, 2
		from OLD_TABLE;
	elsif (TG_OP='UPDATE') then
		update employee_version set end_transaction_id = tx_id 
		where id in (select id from NEW_TABLE);

		insert into employee_version (id, name, salary, transaction_id, operation_type)
		select id, name, salary, tx_id, 1
		from NEW_TABLE;
	elsif (TG_OP='INSERT') then
		insert into employee_version (id, name, salary, transaction_id, operation_type)
		select id, name, salary, tx_id, 0
		from NEW_TABLE;
	end if;
	return null;
end
$$ language plpgsql;


drop  trigger if exists employee_log_trigger_insert on employee;
create trigger employee_log_trigger_insert
after insert on employee 
referencing 
	new table as NEW_TABLE
for each statement EXECUTE function log_employee_changes();

drop  trigger  if exists employee_log_trigger_update on employee;
create trigger employee_log_trigger_update
after update on employee 
referencing 
	old table as OLD_TABLE 
	new table as NEW_TABLE
for each statement EXECUTE function log_employee_changes();

drop  trigger  if exists employee_log_trigger_delete on employee;
create trigger employee_log_trigger_delete
after delete on employee 
referencing 
	old table as OLD_TABLE
for each statement EXECUTE function log_employee_changes();


insert into employee (id, name, salary) 
values (1, 'alice', 110.),
(2, 'bob', 100.),
(3, 'charlie', 99.5);


insert into employee (id, name, salary) 
values (4, 'dave', 939.),
(5, 'eve', 1000.),
(6, 'frida', 909.5);


delete from employee where (salary > 300);
update employee set salary = 200;