PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- HS-4981: catch blocks re-raise with THROW (raiserror cannot re-raise severity > 18 and masks the original error);'
PRINT N'-- HS-4981: Export_EntitySynchronizeItems commits in batches (bounded lock count on large entities)'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Altering procedure [mig_item].[_Export_EntitySynchronizeItems]...'
GO

CREATE OR ALTER procedure [mig_item].[_Export_EntitySynchronizeItems]
	@jobId int
,	@jobInstanceId int
,	@entityID uniqueidentifier
,	@newCount int output
,	@obsoleteCount int output
as
	set nocount on

	-- make sure the expected #sync temp table exists
	if object_id('tempdb..#sync') is null
	begin
		raiserror('#sync temp table not found. Call [mig_item].[Export_EntitySynchronizeItems] first.', 16, 1)
		return
	end

	declare @time datetime2(7) = sysdatetime()

	-- Batched processing: each batch commits its own transaction so the lock count stays
	-- bounded on large entities (HS-4981; 1M+ items in a single transaction exhausted lock
	-- memory). A failed run leaves completed batches in place; a rerun reconciles from the
	-- ExportKey anti-joins below. The caller must NOT hold an enclosing transaction, or the
	-- per-batch commits are nested no-ops and the lock count is unbounded again.
	declare @batchSize int = 50000

	--------------------------------------------------------------------------------------------------------------
	-- Process New items
	--------------------------------------------------------------------------------------------------------------
	drop table if exists #newExport
	create table #newExport (
		[RowNum] int identity(1, 1) not null primary key
	,	[ExportID] uniqueidentifier not null unique
	,   [BreakFields] nvarchar(4000) not null
	)

	-- Determine new items; [RowNum] (identity, arrival order) numbers them for batching
	-- without the sort a row_number() would need
	insert #newExport (
		[ExportID]
	,	[BreakFields]
	)
	select
		A.[ExportID]
	,	A.[BreakFields]
	from
		#sync as A
		left outer join [mig_item].[ExportKey] as B
		on
			B.[EntityID] = @entityID
		and	B.[ExportID] = A.[ExportID]
		where
			B.[ExportID] is null

	-- Return the number of new items
	set @newCount = @@rowcount

	drop table if exists #mapExportIDToItemId
	create table #mapExportIDToItemId (
		[ExportID] uniqueidentifier not null primary key
	,   [ItemId] bigint not null
	)

	-- Loop bound on max(RowNum), not @newCount: identity values are dense for a single
	-- insert, but a gap must never silently skip the tail of the set
	declare @maxRow int = isnull((select max([RowNum]) from #newExport), 0)
	declare @from int = 1
	declare @to int

	while @from <= @maxRow
	begin
		set @to = @from + @batchSize - 1

		-- The map only ever holds the current batch; the joins below are batch-restricted through it
		truncate table #mapExportIDToItemId

		begin transaction

			-- Generate ItemIds
			-- Using MERGE to be able to reference src.[ExportID] in the OUTPUT clause
			merge [mig_item].[Id] with (rowlock) as tgt
			using (
				select [ExportID]
				from #newExport
				where [RowNum] between @from and @to
			) as src
			on 1 = 0 -- force insert for every source row
			when not matched by target then
				insert (
					[CreationDate]
				) values (
					@time
				)
				output
					src.[ExportID]
				,	inserted.[ItemId]
				into #mapExportIDToItemId (
					[ExportID]
				,	[ItemId]
				);

			-- Match ExportIDs and generated ItemIds and insert on ExportKey
			insert [mig_item].[ExportKey] with (RowLock) (
				[ExportID]
			,	[ItemId]
			,	[EntityID]
			,	[BreakFields]
			,	[JobId]
			,	[JobInstanceId]
			)
			select
				A.[ExportID]
			,	B.[ItemId]
			,	@entityID
			,	A.[BreakFields]
			,	@jobId
			,	@jobInstanceId
			from
				#newExport as A
				inner join #mapExportIDToItemId as B
				on
					A.[ExportID] = B.[ExportID]

			-- Normalize key fields for faster searches
			insert [mig_item].[ExportKeyField] (
				[ItemId]
			,	[KeyName]
			,	[KeyValue]
			)
			select
				m.[ItemId]
			,	k.N.value('local-name(.)', 'nvarchar(256)')
			,	k.N.value('(@value)[1]', 'nvarchar(512)')
			from
				#newExport as e
				inner join #mapExportIDToItemId as m
				on
					e.[ExportID] = m.[ExportID]
				cross apply (select cast(e.[BreakFields] as xml) as keyXml) as x
				cross apply x.keyXml.nodes('/*/*') as k(N);

		commit transaction

		set @from = @to + 1
	end

	--------------------------------------------------------------------------------------------------------------
	-- Process Obsolete items
	--------------------------------------------------------------------------------------------------------------
	declare @obsoleteItems [mig_garbage].[InvalidateSet]
	declare @batchCount int

	set @obsoleteCount = 0

	while 1 = 1
	begin
		delete @obsoleteItems

		begin transaction

			-- Determine and remove one batch of obsolete items
			delete top (@batchSize) A
			output deleted.[ItemId] into @obsoleteItems([ItemId])
			from
				[mig_item].[ExportKey] as A with (RowLock)
				left outer join #sync as B
				on
					A.[ExportID] = B.[ExportID]
			where
				A.[EntityID] = @entityID
			and	B.[ExportID] is null

			set @batchCount = @@rowcount
			set @obsoleteCount += @batchCount

			if @batchCount > 0
			begin
				-- Delete normalized key fields for obsolete items
				delete A
				from
					[mig_item].[ExportKeyField] as A with (RowLock)
					inner join @obsoleteItems as B
					on
						A.[ItemId] = B.[ItemId]

				-- Mark as garbage
				exec [mig_garbage].[_SetItemSetObsolete] @obsoleteItems
			end

		commit transaction

		if @batchCount < @batchSize
			break
	end
GO

PRINT N'Altering procedure [mig_item].[Export_EntitySynchronizeItems]...'
GO

CREATE OR ALTER procedure [mig_item].[Export_EntitySynchronizeItems]
	@jobId int
,	@jobInstanceId int
,	@entityID	uniqueidentifier
,	@exportDb varchar(50)
,	@totalCount int output
,	@newCount int output
,	@obsoleteCount int output
as
	set xact_abort on;

	----------------------------------------------------------------------------------------------------------------
	-- Dynamically call the Synchronize procedure in the Export DB and retrieve the sync results in temp table #sync
	----------------------------------------------------------------------------------------------------------------
	declare @sql nvarchar(max) = N'exec ' + @exportDb + N'.[mig_director].[EntityKeyList] @entityId;'

	-- Execute the call
	drop table if exists #sync;
	create table #sync (
		[ExportID] uniqueidentifier primary key
	,	[BreakFields] nvarchar(1024) not null	
	);
	insert into #sync (
		[ExportID]
	,   [BreakFields]
	)
	exec sp_executesql
		@sql
	,   N'@entityId uniqueidentifier'
	,   @entityId = @entityID

	-- Return the total count
	set @totalCount = @@rowcount

	declare @now datetime2(7) = sysdatetime()

	begin try
		-- No enclosing transaction: the synchronize commits in batches internally to keep
		-- the lock count bounded on large entities (HS-4981). A failure leaves completed
		-- batches in place; a rerun reconciles them.
		-- The #sync table is expected by the called procedure
		exec [mig_item].[_Export_EntitySynchronizeItems] @jobId, @jobInstanceId, @entityID, @newCount output, @obsoleteCount output

		-- Update the ExportEntity LastSynchronizeTime (only after a fully completed synchronize)
		update [mig_director].[ExportEntity]
		set
			[LastSynchronizeTime] = @now
		where
			[EntityId] = @entityID
	end try
	begin catch
		if @@trancount > 0 rollback transaction;
		throw;
	end catch;
GO

PRINT N'Altering procedure [mig_item].[_GetKeyId]...'
GO

CREATE OR ALTER procedure [mig_item].[_GetKeyId]
	@businessEntityName varchar(50)
,	@keyValue nvarchar(1024)
,	@maxDeadlockRetries int
,	@keyId bigint output
as
	set nocount on

	declare @retryCount int = 0
	declare @startTranCount int = @@trancount;
	declare @localTransaction bit = 0;

	lblRetry:
	select	
		@keyId = [KeyId]
	from
		[mig_item].[MainKey]
	where
		[BusinessEntityName] = @businessEntityName
	and	[KeyValue] = @keyValue

	if @keyId is not null
		return

	-- Guarded insert: may race with other sessions inserting the same key
	begin try
		-- Try insert first

		-- Only manage a transaction if there wasn't one already
		if 0 = @startTranCount
		begin
			begin transaction;
			set @localTransaction = 1;
		end

		insert [mig_item].[MainKey] (
			[BusinessEntityName] 
		,	[KeyValue]
		) 
		values (
			@businessEntityName 
		,	@keyValue 
		)
		
		set @keyId = scope_identity()

		-- Normalize key fields for faster searches
		insert [mig_item].[MainKeyField] (
			[KeyId]
		,	[KeyName]
		,	[KeyValue]
		)
		select
			@keyId
		,	k.N.value('local-name(.)', 'nvarchar(256)')
		,	k.N.value('(@value)[1]', 'nvarchar(512)')
		from
			(select cast(@keyValue as xml) as keyXml) as x
			cross apply x.keyXml.nodes('/*/*') as k(N);

		-- If we started a local transaction, commit it
		if @localTransaction = 1 and @@trancount > 0
			commit transaction;

		return
	end try
	begin catch
		-- Insert failed

		-- If we started a local transaction, roll it back
		if @localTransaction = 1 and @@trancount > 0
			rollback transaction;

		declare @err int = error_number()

		-- 2601 Duplicate in unique index for BusinessEntityName and KeyValue
		-- 2627 Unique constraint violation
		if @err in (2601, 2627)
		begin
			-- Another session inserted the same key between our SELECT and INSERT
			-- Go back and do the select again
			goto lblRetry
		end
		-- 1205: Transaction (Process ID %d) was deadlocked on %.*ls resources with another process and has been chosen as the deadlock victim. Rerun the transaction.
		-- 1222: Lock request time out period exceeded.
		else if @err in (1205, 1222)
		begin
			if @retryCount < @maxDeadlockRetries 
			begin
				declare @delay time = case @retryCount
					when 0 then '00:00:00.000'
					when 1 then '00:00:00.005'
					when 2 then '00:00:00.010'
					else '00:00:00.020'
				end

				waitfor delay @delay

				set @retryCount = @retryCount + 1

				goto lblRetry
			end
		end;

		-- Any other error or out of retries: Rethrow
		-- Bare throw: raiserror cannot re-raise severity > 18 (HS-4981)
		throw;
	end catch
GO
