CREATE OR REPLACE FUNCTION sch_chameleon.fn_replay_mysql(integer,integer,boolean)
RETURNS sch_chameleon.ty_replay_status AS
$BODY$
	DECLARE
		p_i_max_events		ALIAS FOR $1;
		p_i_id_source		ALIAS FOR $2;
		p_b_exit_on_error	ALIAS FOR $3;
		v_ty_status		sch_chameleon.ty_replay_status;
		v_r_statements		record;
		v_i_id_batch		bigint;
		v_v_log_table		text;
		v_t_ddl			text;
		v_t_main_sql		text;
		v_t_delete_sql		text;
		v_i_replayed		integer;
		v_i_skipped		integer;
		v_i_ddl			integer;
		v_i_evt_replay		bigint[];
		v_i_evt_queue		bigint[];
		v_ts_evt_source		timestamp without time zone;
		v_tab_enabled		boolean;

	BEGIN
		v_i_replayed:=0;
		v_i_ddl:=0;
		v_i_skipped:=0;
		v_ty_status.b_continue:=FALSE;
		v_ty_status.b_error:=FALSE;		
		RAISE DEBUG 'Searching batches to replay for source id: %', p_i_id_source;
		v_i_id_batch:= (
			SELECT 
				bat.i_id_batch 
			FROM 
				sch_chameleon.t_replica_batch bat
				INNER JOIN  sch_chameleon.t_batch_events evt
				ON
					evt.i_id_batch=bat.i_id_batch
			WHERE 
					bat.b_started 
				AND	bat.b_processed 
				AND	NOT bat.b_replayed
				AND	bat.i_id_source=p_i_id_source
			ORDER BY 
				bat.ts_created 
			LIMIT 1
			)
		;

		v_v_log_table:=(
			SELECT 
				v_log_table
			FROM 
				sch_chameleon.t_replica_batch 
			WHERE 
				i_id_batch=v_i_id_batch
			)
		;
		IF v_i_id_batch IS NULL 
		THEN
			RAISE DEBUG 'There are no batches available for replay';
			RETURN v_ty_status;
		END IF;
		
		RAISE DEBUG 'Found id_batch %, data in log table %', v_i_id_batch,v_v_log_table;
		RAISE DEBUG 'Building a list of event id with max length %...', p_i_max_events;
		v_i_evt_replay:=(
			SELECT 
				i_id_event[1:p_i_max_events] 
			FROM 
				sch_chameleon.t_batch_events 
			WHERE 
				i_id_batch=v_i_id_batch
		);
		
		
		v_i_evt_queue:=(
			SELECT 
				i_id_event[p_i_max_events+1:array_length(i_id_event,1)] 
			FROM 
				sch_chameleon.t_batch_events 
			WHERE 
				i_id_batch=v_i_id_batch
		);
		
		RAISE DEBUG 'Finding the last executed event''s timestamp...';
		v_ts_evt_source:=(
			SELECT 
				to_timestamp(i_my_event_time)
			FROM	
				sch_chameleon.t_log_replica
			WHERE
					i_id_event=v_i_evt_replay[array_length(v_i_evt_replay,1)]
				AND	i_id_batch=v_i_id_batch
		);
		
		RAISE DEBUG 'Generating the main loop sql';

		v_t_main_sql:=format('
			SELECT 
				i_id_event AS i_id_event,
				enm_binlog_event,
				(enm_binlog_event=''ddl'')::integer as i_ddl,
				(enm_binlog_event<>''ddl'')::integer as i_replay,
				t_binlog_name,
				i_binlog_position,
				v_table_name,
				v_schema_name,
				t_pk_data,
				CASE
					WHEN enm_binlog_event = ''ddl''
					THEN 
						t_query
					WHEN enm_binlog_event = ''insert''
					THEN
						format(
							''INSERT INTO %%I.%%I %%s;'',
							v_schema_name,
							v_table_name,
							t_dec_data
							
						)
					WHEN enm_binlog_event = ''update''
					THEN
						format(
							''UPDATE %%I.%%I SET %%s WHERE %%s;'',
							v_schema_name,
							v_table_name,
							t_dec_data,
							t_pk_data
						)
					WHEN enm_binlog_event = ''delete''
					THEN
						format(
							''DELETE FROM %%I.%%I WHERE %%s;'',
							v_schema_name,
							v_table_name,
							t_pk_data
						)
					
				END AS t_sql
			FROM 
			(
				SELECT 
					pk.i_id_event,
					pk.v_table_name,
					pk.v_schema_name,
					pk.enm_binlog_event,
					pk.t_binlog_name,
					pk.i_binlog_position,
					pk.t_query as t_query,
					pk.ts_event_datetime,
					pk.t_dec_data,
					string_agg(DISTINCT 
							CASE
								WHEN pk.v_table_pkey IS NOT NULL
								THEN
									format(
										''%%I=%%L'',
										pk.v_table_pkey,
										CASE 
											WHEN pk.enm_binlog_event = ''update''
											THEN
												pk.jsb_event_before->>v_table_pkey
											ELSE
												pk.jsb_event_after->>v_table_pkey
										END 	
								
									)
							END
					,'' AND '') as  t_pk_data
					
				FROM
				(
					SELECT 
						dec.i_id_event,
						dec.v_table_name,
						dec.v_schema_name,
						dec.enm_binlog_event,
						dec.t_binlog_name,
						dec.i_binlog_position,
						dec.t_query as t_query,
						dec.ts_event_datetime,
						CASE
							WHEN dec.enm_binlog_event = ''insert''
							THEN
							format(''(%%s) VALUES (%%s)'',string_agg(format(''%%I'',dec.t_column),'',''),string_agg(format(''%%L'',dec.jsb_event_after->>t_column),'',''))
							WHEN dec.enm_binlog_event = ''update''
							THEN
								string_agg(format(''%%I=%%L'',dec.t_column,dec.jsb_event_after->>t_column),'','')
							
						END AS t_dec_data,
						unnest(v_table_pkey) as v_table_pkey,
						dec.jsb_event_after,
						dec.jsb_event_before
						
					FROM 
					(
						SELECT --DISTINCT ON (t_column)
							log.i_id_event,
							log.v_table_name,
							log.v_schema_name,
							log.enm_binlog_event,
							log.t_binlog_name,
							log.i_binlog_position,
							coalesce(log.jsb_event_after,''{"foo":"bar"}''::jsonb) as jsb_event_after,
							(jsonb_each_text(coalesce(log.jsb_event_after,''{"foo":"bar"}''::jsonb))).key AS t_column,
							log.jsb_event_before,
							log.t_query as t_query,
							log.ts_event_datetime,
							v_table_pkey
						FROM 
							sch_chameleon.%I log
							INNER JOIN sch_chameleon.t_replica_tables tab
								ON 
										tab.v_table_name=log.v_table_name
									AND	tab.v_schema_name=log.v_schema_name
						WHERE
								tab.b_replica_enabled
							AND	i_id_event = ANY(%L)
						
					) dec
					GROUP BY 
						dec.i_id_event,
						dec.v_table_name,
						dec.v_schema_name,
						dec.enm_binlog_event,
						dec.t_query,
						dec.ts_event_datetime,
						dec.t_binlog_name,
						dec.i_binlog_position,
						dec.v_table_pkey,
						dec.jsb_event_after,
						dec.jsb_event_before
					) pk
				GROUP BY 
					pk.i_id_event,
					pk.v_table_name,
					pk.v_schema_name,
					pk.enm_binlog_event,
					pk.t_binlog_name,
					pk.i_binlog_position,
					pk.t_query,
					pk.ts_event_datetime,
					pk.t_dec_data
				
					
			) par
			ORDER BY 
				i_id_event ASC
			;		
		',v_v_log_table,v_i_evt_replay);
		RAISE DEBUG '%',v_t_main_sql;
		FOR v_r_statements IN EXECUTE v_t_main_sql
		LOOP
			
			BEGIN
				EXECUTE v_r_statements.t_sql;
				v_i_ddl:=v_i_ddl+v_r_statements.i_ddl;
				v_i_replayed:=v_i_replayed+v_r_statements.i_replay;
				
				
			EXCEPTION
				WHEN OTHERS
				THEN
				RAISE NOTICE 'An error occurred when replaying data for the table %.%',v_r_statements.v_schema_name,v_r_statements.v_table_name;
				RAISE NOTICE 'SQLSTATE: % - ERROR MESSAGE %',SQLSTATE, SQLERRM;
				RAISE DEBUG 'SQL EXECUTED: % ',v_r_statements.t_sql;
				RAISE NOTICE 'The table %.% has been removed from the replica',v_r_statements.v_schema_name,v_r_statements.v_table_name;
				v_ty_status.v_table_error:=array_append(v_ty_status.v_table_error, format('%I.%I SQLSTATE: %s - ERROR MESSAGE: %s',v_r_statements.v_schema_name,v_r_statements.v_table_name,SQLSTATE, SQLERRM)::character varying) ;
				RAISE NOTICE 'Adding error log entry for table %.% ',v_r_statements.v_schema_name,v_r_statements.v_table_name;
				INSERT INTO sch_chameleon.t_error_log
							(
								i_id_batch, 
								i_id_source,
								v_schema_name, 
								v_table_name, 
								t_table_pkey, 
								t_binlog_name, 
								i_binlog_position, 
								ts_error, 
								t_sql,
								t_error_message
							)
							SELECT 
								i_id_batch, 
								p_i_id_source,
								v_schema_name, 
								v_table_name, 
								v_r_statements.t_pk_data as t_table_pkey, 
								t_binlog_name, 
								i_binlog_position, 
								clock_timestamp(), 
								quote_literal(v_r_statements.t_sql) as t_sql,
								format('%s - %s',SQLSTATE, SQLERRM) as t_error_message
							FROM
								sch_chameleon.t_log_replica  log
							WHERE 
								log.i_id_event=v_r_statements.i_id_event
						;
				IF p_b_exit_on_error
				THEN
					v_ty_status.b_continue:=FALSE;
					v_ty_status.b_error:=TRUE;
					RETURN v_ty_status;
				ELSE
				
					RAISE NOTICE 'Statement %', v_r_statements.t_sql;
					UPDATE sch_chameleon.t_replica_tables 
						SET 
							b_replica_enabled=FALSE
					WHERE
							v_schema_name=v_r_statements.v_schema_name
						AND	v_table_name=v_r_statements.v_table_name
					;

					RAISE NOTICE 'Deleting the log entries for the table %.% ',v_r_statements.v_schema_name,v_r_statements.v_table_name;
					DELETE FROM sch_chameleon.t_log_replica  log
					WHERE
							v_table_name=v_r_statements.v_table_name
						AND	v_schema_name=v_r_statements.v_schema_name
						AND 	i_id_batch=v_i_id_batch
					;
				END IF;
			END;
		END LOOP;
		IF v_ts_evt_source IS NOT NULL
		THEN
			UPDATE sch_chameleon.t_last_replayed
				SET
					ts_last_replayed=v_ts_evt_source
			WHERE 	
				i_id_source=p_i_id_source
			;
		END IF;
		IF v_i_replayed=0 AND v_i_ddl=0
		THEN
			DELETE FROM sch_chameleon.t_log_replica
			WHERE
    			    i_id_batch=v_i_id_batch
			;
				
			GET DIAGNOSTICS v_i_skipped = ROW_COUNT;
			RAISE DEBUG 'SKIPPED ROWS: % ',v_i_skipped;

			UPDATE ONLY sch_chameleon.t_replica_batch  
			SET 
				b_replayed=True,
				i_skipped=v_i_skipped,
				ts_replayed=clock_timestamp()
				
			WHERE
				i_id_batch=v_i_id_batch
			;

			DELETE FROM sch_chameleon.t_batch_events
			WHERE
				i_id_batch=v_i_id_batch
			;

			v_ty_status.b_continue:=FALSE;
		ELSE
			UPDATE ONLY sch_chameleon.t_replica_batch  
			SET 
				i_ddl=coalesce(i_ddl,0)+v_i_ddl,
				i_replayed=coalesce(i_replayed,0)+v_i_replayed,
				i_skipped=v_i_skipped,
				ts_replayed=clock_timestamp()
				
			WHERE
				i_id_batch=v_i_id_batch
			;

			UPDATE sch_chameleon.t_batch_events
				SET
					i_id_event = v_i_evt_queue
			WHERE
				i_id_batch=v_i_id_batch
			;

			DELETE FROM sch_chameleon.t_log_replica
			WHERE
					i_id_batch=v_i_id_batch
				AND 	i_id_event=ANY(v_i_evt_replay) 
			;
			v_ty_status.b_continue:=TRUE;
			RETURN v_ty_status;
		END IF;
		v_i_id_batch:= (
			SELECT 
				bat.i_id_batch 
			FROM 
				sch_chameleon.t_replica_batch bat
				INNER JOIN  sch_chameleon.t_batch_events evt
				ON
					evt.i_id_batch=bat.i_id_batch
			WHERE 
					bat.b_started 
				AND	bat.b_processed 
				AND	NOT bat.b_replayed
				AND	bat.i_id_source=p_i_id_source
			ORDER BY 
				bat.ts_created 
			LIMIT 1
			)
		;
		
		IF v_i_id_batch IS NOT NULL
		THEN
			v_ty_status.b_continue:=TRUE;
		END IF;
		
		
		RETURN v_ty_status;

		
		
	END;
	
$BODY$
LANGUAGE plpgsql;