USE Credit;
GO

IF OBJECT_ID('cdc_lsn_tracking') IS NOT NULL
BEGIN
	DROP TABLE dbo.cdc_lsn_tracking;
END
GO

CREATE TABLE dbo.cdc_lsn_tracking
( capture_instance sysname,
  last_processed_lsn binary(10) )
GO

CREATE UNIQUE CLUSTERED INDEX UX_cdc_lsn_tracking ON dbo.cdc_lsn_tracking (capture_instance)
GO


IF OBJECT_ID('Process_CDC_changes') IS NOT NULL
BEGIN
	DROP PROCEDURE dbo.Process_CDC_changes;
END
GO

-- Process using dynamic sql for more flexibility
CREATE PROCEDURE dbo.Process_CDC_changes (@capture_instance sysname, @row_filter_option sysname, @mark_processed bit = 0)
AS
BEGIN

	IF NOT EXISTS (SELECT * FROM cdc.change_tables WHERE capture_instance = @capture_instance)
	BEGIN
		RAISERROR('The specified capture_instance is valid for the current cdc_configuration', 16, 1);
		RETURN;
	END

	IF NOT EXISTS (SELECT capture_instance FROM dbo.cdc_lsn_tracking WHERE capture_instance = @capture_instance)
	BEGIN
		INSERT INTO dbo.cdc_lsn_tracking (capture_instance) VALUES (@capture_instance);
	END

	-- Get the current min LSN for the capture instance based on datetime
	DECLARE	@from_lsn binary(10),
			@to_lsn binary(10);

	-- Get the current min LSN for the capture instance based on datetime
	SELECT	@from_lsn =  ISNULL(last_processed_lsn, sys.fn_cdc_get_min_lsn(@capture_instance)),
			@to_lsn =  sys.fn_cdc_decrement_lsn(sys.fn_cdc_get_max_lsn())
	FROM dbo.cdc_lsn_tracking
	WHERE capture_instance = @capture_instance;

	-- If a DDL change has occured, consider stopping at the ddl_lsn value so you don't miss data
	IF EXISTS (SELECT * 
				FROM [cdc].[ddl_history] AS d 
				INNER JOIN [cdc].[change_tables] AS ct 
					ON d.object_id = ct.object_id and 
						d.source_object_id = ct.source_object_id
				WHERE capture_instance = @capture_instance)
		SELECT @to_lsn = [ddl_lsn]
		FROM [cdc].[ddl_history] AS d 
		INNER JOIN [cdc].[change_tables] AS ct 
			ON d.object_id = ct.object_id and 
				d.source_object_id = ct.source_object_id
		WHERE capture_instance = @capture_instance

	DECLARE @sqlcmd NVARCHAR(MAX) = '
		-- Get all changes fro the capture instance within the LSN range
		SELECT 
			CASE [__$operation] 
				WHEN 1 THEN ''Delete''
				WHEN 2 THEN ''INSERT''
				WHEN 3 THEN ''Update From Value''
				WHEN 4 THEN ''Update To Value''
			END AS [operation],
			sys.fn_cdc_map_lsn_to_time([__$start_lsn]) AS [trans_time],
			*
		FROM cdc.fn_cdc_get_all_changes_' +@capture_instance+' (@from_lsn, @to_lsn, @row_filter_option);'

	DECLARE @params NVARCHAR(MAX) = N'@from_lsn binary(10), @to_lsn binary(10), @row_filter_option sysname';

	EXECUTE sp_executesql 
		@sqlcmd,
		@params,
		@from_lsn = @from_lsn,
		@to_lsn = @to_lsn,
		@row_filter_option = @row_filter_option;

	IF @mark_processed = 1
	BEGIN
		-- Update the lsn tracking table to reflect the last processed transaction lsn from the max LSN
		UPDATE dbo.cdc_lsn_tracking
		SET last_processed_lsn = @to_lsn
		WHERE capture_instance = @capture_instance;
	END
END
GO

-- Perform an update to one of the captured charges
UPDATE dbo.charge
SET charge_amt = 4622.00
WHERE charge_no = 2000001;
GO

EXECUTE Process_CDC_changes @capture_instance = 'dbo_charge', @row_filter_option = 'all update old', @mark_processed = 1;

-- Perform an update to one of the captured charges
UPDATE dbo.charge
SET charge_amt = 4621.00
WHERE charge_no = 2000001;
GO

EXECUTE Process_CDC_changes @capture_instance = 'dbo_charge', @row_filter_option = 'all update old', @mark_processed = 1;

