create database mds;

SET GLOBAL log_bin_trust_function_creators = 1;
USE mds;

DROP FUNCTION IF EXISTS get_gtid_to_skip;
DELIMITER |
CREATE FUNCTION get_gtid_to_skip()
RETURNS LONGTEXT
BEGIN
  DECLARE tmp LONGTEXT;
  select replace(replace(regexp_substr(last_error_message, "transaction '.*'"), "transaction ",""), "'", "") `gtid_to_bypass` from performance_schema.replication_applier_status_by_coordinator INTO tmp;
  RETURN tmp;
END|                    
DELIMITER ; 

DROP PROCEDURE IF EXISTS skip_gtid;
DELIMITER |
CREATE PROCEDURE skip_gtid(gtid_skip LONGTEXT)
BEGIN
  SET GTID_NEXT=gtid_skip;
  START TRANSACTION;
  COMMIT;
  SET GTID_NEXT="AUTOMATIC";  
END|                    
DELIMITER ;

DROP PROCEDURE IF EXISTS skip_replication_error;
DELIMITER |
CREATE PROCEDURE skip_replication_error()
BEGIN
  DECLARE gtid_skip LONGTEXT;
  select concat("+", get_gtid_to_skip()) INTO gtid_skip;
  IF(gtid_skip IS NULL) THEN
     select "No replication error to skip" as `message`;
  ELSE
    call sys.set_gtid_purged(gtid_skip);
    select gtid_skip `purged GTID`;
  END IF;
END|                    
DELIMITER ;  

DROP PROCEDURE IF EXISTS skip_replication_error_with_empty_gtid;
DELIMITER |
CREATE PROCEDURE skip_replication_error_with_empty_gtid()
BEGIN
  DECLARE gtid_skip LONGTEXT;
  select get_gtid_to_skip() INTO gtid_skip;
  IF(gtid_skip IS NULL) THEN
     select "No replication error to skip" as `message`;
  ELSE
    select gtid_skip `skipped GTID`;
    SET GTID_NEXT=gtid_skip;
    START TRANSACTION;
    COMMIT;
   SET GTID_NEXT="AUTOMATIC";  
  END IF;
END|                    
DELIMITER ;  


DROP PROCEDURE IF EXISTS replication_status;
DELIMITER |
CREATE PROCEDURE replication_status()
BEGIN

SELECT
  conn_status.channel_name as channel_name,
  conn_status.service_state as IO_thread,
  applier_status.service_state as SQL_thread, 
    if(LAST_QUEUED_TRANSACTION = "ANONYMOUS", "N/A",  
    if(GTID_SUBTRACT(LAST_QUEUED_TRANSACTION, LAST_APPLIED_TRANSACTION) = "","0" ,
      abs(time_to_sec(if(time_to_sec(APPLYING_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP)=0,0,
      timediff(APPLYING_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP,now())))))) `lag_in_sec`                            
  FROM
  performance_schema.replication_connection_status AS conn_status
JOIN performance_schema.replication_applier_status_by_worker AS applier_status
  ON applier_status.channel_name = conn_status.channel_name
  order by 4 desc limit 1
  ;   
END|                    
DELIMITER ; 

DROP PROCEDURE IF EXISTS replication_status_extended;
DELIMITER |
CREATE PROCEDURE replication_status_extended()
BEGIN

SELECT
  conn_status.channel_name as channel_name,
  conn_status.service_state as IO_thread,
  applier_status.service_state as SQL_thread,
  conn_status.LAST_QUEUED_TRANSACTION as last_queued_transaction,
  applier_status.LAST_APPLIED_TRANSACTION as last_applied_transaction,
  if(LAST_APPLIED_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP = 0, 0,
  LAST_APPLIED_TRANSACTION_END_APPLY_TIMESTAMP -
                            LAST_APPLIED_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP) 'rep delay (sec)',                         
  LAST_QUEUED_TRANSACTION_START_QUEUE_TIMESTAMP -
                           LAST_QUEUED_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP 'transport time',
  LAST_QUEUED_TRANSACTION_END_QUEUE_TIMESTAMP -
                           LAST_QUEUED_TRANSACTION_START_QUEUE_TIMESTAMP 'time RL',
  LAST_APPLIED_TRANSACTION_END_APPLY_TIMESTAMP -
                           LAST_APPLIED_TRANSACTION_START_APPLY_TIMESTAMP 'apply time',
  if(LAST_QUEUED_TRANSACTION = "ANONYMOUS", "N/A",
  if(GTID_SUBTRACT(LAST_QUEUED_TRANSACTION, LAST_APPLIED_TRANSACTION) = "","0" ,
      abs(time_to_sec(if(time_to_sec(APPLYING_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP)=0,0,
      timediff(APPLYING_TRANSACTION_ORIGINAL_COMMIT_TIMESTAMP,now())))))) `lag_in_sec`
FROM
  performance_schema.replication_connection_status AS conn_status
JOIN performance_schema.replication_applier_status_by_worker AS applier_status
  ON applier_status.channel_name = conn_status.channel_name
  ;
END|                    
DELIMITER ; 

DROP PROCEDURE IF EXISTS replication_errors;
DELIMITER |
CREATE PROCEDURE replication_errors()
BEGIN
select channel_name, last_error_timestamp, last_error_message from performance_schema.replication_connection_status where last_error_message not like "";
select channel_name, last_error_timestamp, last_error_message from performance_schema.replication_applier_status_by_coordinator where last_error_message not like "";
select channel_name, last_error_timestamp, last_error_message from performance_schema.replication_applier_status_by_worker where last_error_message not like "";
END|                    
DELIMITER ;