PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- EventSystem streaming source SP and TVP-based bridge-back'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Creating procedure [mig_tracking].[EventSystem_List]...'
GO

CREATE OR ALTER PROCEDURE [mig_tracking].[EventSystem_List]
as
begin
	set nocount on

	-----------------------------------------------------
	-- System Events (consumed as a streamed TVP by
	-- [mig_tracking].[EventSystem_InsertIfMissing] in
	-- the Project DB; see DbProject.EventSystem_InsertIfMissing).
	-- [Message] is VARCHAR(256) here; the receiving TVP column
	-- [EventSystemInput].[MessageFmt] is NVARCHAR(MAX). The
	-- DbDataReader -> TVP wire format requires matching column
	-- types, so convert + alias here.
	-----------------------------------------------------
	select distinct
		[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	convert(nvarchar(max), [Message]) as [MessageFmt]
	from [mig_valid].[EventSystem]
end
GO

PRINT N'Altering procedure [mig_tracking].[Actual_List] (drop System Events result set)...'
GO

CREATE OR ALTER procedure [mig_tracking].[Actual_List]
as
begin
	set nocount on

	-----------------------------------------------------
	-- List SmartTexts
	-----------------------------------------------------
	select
		[ItemID]
	,	[Xml]
	from [mig_director].[SmartText]

	-----------------------------------------------------
	--  User Events
	-----------------------------------------------------
	select distinct
		A.[EventId]
	,	A.[EventTextId]
	,	A.[Engine]
	,	A.[Step]
	,	A.[Disposition]
	,	A.[Impact]
	,	A.[Receiver]
	,	A.[EventName]
	,	cast(case
			when B.[ItemID] is null then 1
			else 0
		end as bit) as [TextMissing]
	into #userMessage
	from [mig_valid].[EventUser] as A
	left outer join [mig_director].[SmartText] as B
		on B.[ItemID] = A.[EventTextId]

	select
		[EventId]
	,	[EventTextId]
	,	[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	[EventName]
	from #userMessage

	-----------------------------------------------------
	-- System Events are sourced from
	-- [mig_tracking].[EventSystem_List] and inserted via
	-- [mig_tracking].[EventSystem_InsertIfMissing] on the
	-- Project DB (see DbProject.EventSystem_InsertIfMissing).
	-----------------------------------------------------

	-----------------------------------------------------
	-- List current PartitionValues
	-----------------------------------------------------
	select distinct
		[PartitionValue]
	from
		[mig_valid].[Item]

	-----------------------------------------------------
	-- Report missing smarttexts for User Events
	-----------------------------------------------------
	select
		[EventTextID]
	,	[EventName]
	from #userMessage
	where [TextMissing] = 1
end
GO

PRINT N'Creating type [mig_tracking].[EventSystemMessageInput]...'
GO

IF TYPE_ID(N'[mig_tracking].[EventSystemMessageInput]') IS NULL
	CREATE TYPE [mig_tracking].[EventSystemMessageInput] AS TABLE
	(
		[MessageID]		uniqueidentifier	not null
	,	[Step]			tinyint				not null
	,	[Engine]		tinyint				not null
	,	[Disposition]	tinyint				not null
	,	[Impact]		tinyint				not null
	,	[Receiver]		tinyint				not null
	,	[MessageFmt]	nvarchar(max)		not null
	)
GO

PRINT N'Creating type [mig_tracking].[EventUserMessageInput]...'
GO

IF TYPE_ID(N'[mig_tracking].[EventUserMessageInput]') IS NULL
	CREATE TYPE [mig_tracking].[EventUserMessageInput] AS TABLE
	(
		[MessageID]		uniqueidentifier	not null
	,	[Step]			tinyint				not null
	,	[Engine]		tinyint				not null
	,	[Disposition]	tinyint				not null
	,	[Impact]		tinyint				not null
	,	[Receiver]		tinyint				not null
	,	[EventID]		uniqueidentifier	not null
	)
GO

PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- Delivery statistics: Statistic columns, _Delivery_Statistics UDF, Actual_Update pivot extension'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Altering [mig_tracking].[Statistic] (add [DelDiscarded] column)...'
GO

IF NOT EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'[mig_tracking].[Statistic]') AND name = N'DelDiscarded')
	ALTER TABLE [mig_tracking].[Statistic]
		ADD [DelDiscarded] BIGINT CONSTRAINT [DF_Statistic_DelDiscarded] DEFAULT (0) NOT NULL
GO

PRINT N'Altering [mig_tracking].[Statistic] (add [Delivered] column)...'
GO

IF NOT EXISTS (SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'[mig_tracking].[Statistic]') AND name = N'Delivered')
	ALTER TABLE [mig_tracking].[Statistic]
		ADD [Delivered] BIGINT CONSTRAINT [DF_Statistic_Delivered] DEFAULT (0) NOT NULL
GO

PRINT N'Creating function [mig_director].[_Delivery_Statistics]...'
GO

CREATE OR ALTER FUNCTION [mig_director].[_Delivery_Statistics]()
returns table
as
	return (
		with list as (
			select
				B.[PartitionValue]
			,	C.[EntityID]
			, count(*) as [Count]
			, sum(
					case
						when F.[JobId] is null then 0
						else 1
					end
				) as [Processed]
			, sum(
					case
						when isnull(F.[Discarded], 1) = 0 then 1 -- 0:Delivered
						else 0
					end
				) as [Delivered]
			, sum(
					case
						when isnull(F.[Discarded], 0) = 1 then 1 -- 1:DeliveryDiscarded
						else 0
					end
				) as [Discarded]
			from
				[mig_item].[Id] as A with (NoLock)
				inner join [mig_item].[Item] as B with (NoLock)
					on	A.[ItemId] = B.[ItemId]
				inner join [mig_item].[ExportKey] as C with (NoLock)
					on	A.[ItemId] = C.[ItemId]
				inner join [mig_item].[ExportItem] as D with (NoLock)
					on	A.[ItemId] = D.[ItemId] and D.[Result] = 0 -- 0:Exported
				inner join [mig_item].[ImportItem] as E with (NoLock)
					on	A.[ItemId] = E.[ItemId] and E.[Result] = 2 and E.[JobId] >= D.[JobId] -- 2:Success
				left outer join [mig_item].[FeedbackItem] as F with (NoLock)
					on	A.[ItemId] = F.[ItemId] and F.[JobId] >= E.[JobId]
			where
				A.[InvalidatedTime] is null
			group by
				B.[PartitionValue]
			,	C.[EntityID]
		)

		select
			C.[PartitionValue]
		,	A.[EntityID]
		,	B.[BusinessEntityName]
		, A.[EntityName]
		, C.[Count]
		, C.[Delivered]
		, C.[Discarded]
		, C.[Count] - C.[Processed] as [Difference]
		from
			[mig_director].[ExportEntity] as A with (NoLock)
			inner join [mig_director].[BusinessEntity] as B with (NoLock)
				on A.[BusinessEntityID] = B.[BusinessEntityID]
			left outer join list as C
				on A.[EntityID] = C.[EntityID]
		where
			A.[IsPresent] = 1 and A.[Ignored] = 0
	)
GO

PRINT N'Altering procedure [mig_tracking].[Actual_Update] (TVP parameters + Delivery pivot)...'
GO

CREATE OR ALTER procedure [mig_tracking].[Actual_Update]
	@systemMessages [mig_tracking].[EventSystemMessageInput] readonly
,   @userMessages [mig_tracking].[EventUserMessageInput] readonly
,	@entitiesToProcess varchar(max) = null
,	@runDate datetime2
,   @user varchar(256)
,   @jobInstanceId int
as
begin
	set nocount on;

	declare @start datetime = getdate()
	declare @logId bigint
	declare @text varchar(max)  =
'------------------------------------------------------------------
Elapsed  [mig_tracking].[Actual_Update]
------------------------------------------------------------------'

	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 0, @text, @logId output

	-- Table to hold EntityID to process
	declare @entities table(EntityID uniqueidentifier, primary key (EntityID))

	if @entitiesToProcess is not null
	begin
		insert into @entities
		select distinct [Id] from [mig_utility].[ParseGuids](@entitiesToProcess, ',')
	end

	-----------------------------------------------------
	-- EventParameter
	-----------------------------------------------------
	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Insert EventParameter', @logId output

	;with
		usrEv as (select distinct [ParameterHash], [GroupHash], [ParameterXml] from [mig_valid].[EventUser])
	,	sysEv as (select distinct [ParameterHash], [GroupHash], [ParameterXml] from [mig_valid].[EventSystem])
	, allEv as (
		select * from usrEv
		union
		select * from sysEv
	)
	insert [mig_tracking].[EventParameter] (
		[HashValue]
	,	[GroupHash]
	,   [ParameterXml]
	,   [RunDate]
	)
	select
		A.[ParameterHash]
	,	A.[GroupHash]
	,   A.[ParameterXml]
	,   @runDate
	from allEv as A
	left outer join [mig_tracking].[EventParameter] as B
	on
		B.[HashValue] = A.[ParameterHash]
	where
		B.[HashValue] is null

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	-----------------------------------------------------
	-- RootItemEvent
	-----------------------------------------------------
	if @entitiesToProcess is null
	begin
		if exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent_EntityID')
			drop index [IX_RootItemEvent_EntityID] on [mig_tracking].[RootItemEvent]
		if exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent')
			drop index [IX_RootItemEvent] on [mig_tracking].[RootItemEvent]
		if exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent_ItemId')
			drop index [IX_RootItemEvent_ItemId] on [mig_tracking].[RootItemEvent]

		truncate table [mig_tracking].[RootItemEvent]
	end
	else
	begin
		exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Delete RootItemEvent', @logId output
		delete [mig_tracking].[RootItemEvent]
		from
			[mig_tracking].[RootItemEvent] as A
			inner join @entities as B
			on
				A.[EntityID] = B.[EntityID]
		exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0
	end

	-- User events

	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Insert RootItemEvent (User)', @logId output

	;with
		[events] as (
			select
				B.[EntityID]
			,	C.[PartitionValue]
			,	A.[ItemId]
			,	A.[Step]
			,	A.[Engine]
			,	A.[Receiver]
			,	A.[Disposition]
			,	A.[Impact]
			,	A.[EventID]
			,	A.[ParameterHash]
			,	A.[GroupHash]
			,	A.[ParameterXml]
			,	count(*) AS [EventCount]
			from
				[mig_valid].[EventUser] as A
				inner join [mig_item].[ExportKey] as B
				on
					A.[ItemId] = B.[ItemId]
				inner join [mig_item].[Item] as C
				on
					A.[ItemId] = C.[ItemId]
			where
				(@entitiesToProcess is null or B.[EntityID] in (select [EntityID] from @entities))
			group by
				B.[EntityID]
			,	C.[PartitionValue]
			,	A.[ItemId]
			,	A.[Engine]
			,	A.[Step]
			,	A.[Receiver]
			,	A.[Disposition]
			,	A.[Impact]
			,	A.[EventID]
			,	A.[ParameterHash]
			,	A.[GroupHash]
			,	A.[ParameterXml]
		)
	insert [mig_tracking].[RootItemEvent] (
		[RowID]
	,	[EntityID]
	,	[ItemId]
	,	[PartitionValue]
	,	[MessageID]
	,	[ParameterHash]
	,	[GroupHash]
	,	[EventCount]
	,	[RunDate])
	select
		newid() -- To detect problem with join
	,	A.[EntityID]
	,	A.[ItemId]
	,	A.[PartitionValue]
	,	B.[MessageID]
	,	A.[ParameterHash]
	,	A.[GroupHash]
	,	A.[EventCount]
	,	@runDate
	from
		[events] as A
		inner join @userMessages as B
		on
			B.[Step] = A.[Step]
		and	B.[Engine] = A.[Engine]
		and	B.[Disposition] = A.[Disposition]
		and	B.[Impact] = A.[Impact]
		and	B.[Receiver] = A.[Receiver]
		and	B.[EventID] = A.[EventID]

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	-- System events

	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Insert RootItemEvent (System)', @logId output

	;with
		[events] as (
			select
				B.[EntityID]
			,	C.[PartitionValue]
			,	A.[ItemId]
			,	A.[Step]
			,	A.[Engine]
			,	A.[Receiver]
			,	A.[Disposition]
			,	A.[Impact]
			,	A.[Message]
			,	A.[ParameterHash]
			,	A.[GroupHash]
			,	A.[ParameterXml]
			,	count(*) AS [EventCount]
			from
				[mig_valid].[EventSystem] as A
				inner join [mig_item].[ExportKey] as B
				on
					A.[ItemId] = B.[ItemId]
				inner join [mig_item].[Item] as C
				on
					A.[ItemId] = C.[ItemId]
			where
				(@entitiesToProcess is null or B.[EntityID] in (select [EntityID] from @entities))
			group by
				B.[EntityID]
			,	C.[PartitionValue]
			,	A.[ItemId]
			,	A.[Engine]
			,	A.[Step]
			,	A.[Receiver]
			,	A.[Disposition]
			,	A.[Impact]
			,	A.[Message]
			,	A.[ParameterHash]
			,	A.[GroupHash]
			,	A.[ParameterXml]
		)
	insert [mig_tracking].[RootItemEvent] (
		[RowID]
	,	[EntityID]
	,	[ItemId]
	,	[PartitionValue]
	,	[MessageID]
	,	[ParameterHash]
	,	[GroupHash]
	,	[EventCount]
	,	[RunDate])
	select
		newid() -- To detect problem with join
	,	A.[EntityID]
	,	A.[ItemId]
	,	A.[PartitionValue]
	,	B.[MessageID]
	,	A.[ParameterHash]
	,	A.[GroupHash]
	,	A.[EventCount]
	,	@runDate
	from
		[events] as A
		inner join @systemMessages as B
		on
			B.[Step] = A.[Step]
		and	B.[Engine] = A.[Engine]
		and	B.[Disposition] = A.[Disposition]
		and	B.[Impact] = A.[Impact]
		and	B.[Receiver] = A.[Receiver]
		and	B.[MessageFmt] = A.[Message]

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	if not exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent_EntityID')
	begin
		exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Create EntityID index on RootItemEvent', @logId output
		create nonclustered index [IX_RootItemEvent_EntityID] on [mig_tracking].[RootItemEvent]
		(
			[EntityID] asc
		)
		exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0
	end

	if not exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent')
	begin
		exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Create RootItemEvent index on RootItemEvent', @logId output
		create nonclustered index [IX_RootItemEvent] on [mig_tracking].[RootItemEvent]
		(
			[MessageID] asc,
			[EntityID] asc,
			[PartitionValue] asc,
			[ItemId] asc
		) include (
			[ParameterHash]
		)
		exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0
	end

	if not exists (select * from sys.indexes where object_id = object_id(N'[mig_tracking].[RootItemEvent]') and name = N'IX_RootItemEvent_ItemId')
	begin
		exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Create RootItemEvent_ItemId index on RootItemEvent', @logId output
		create nonclustered index [IX_RootItemEvent_ItemId] on [mig_tracking].[RootItemEvent]
		(
			[ItemId] asc
		)
		exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0
	end

	-----------------------------------------------------
	-- RootEntityEvent
	-----------------------------------------------------
	if @entitiesToProcess is null
	begin
		truncate table [mig_tracking].[RootEntityEvent];
	end
	else
	begin
		exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Delete RootEntityEvent', @logId output
		delete [mig_tracking].[RootEntityEvent]
		from
			[mig_tracking].[RootEntityEvent] as A
			inner join @entities as B
			on
				A.[EntityID] = B.[EntityID]
		exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0
	end

	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Insert RootEntityEvent', @logId output

	insert [mig_tracking].[RootEntityEvent] (
		[EntityID]
	,	[PartitionValue]
	,	[MessageID]
	,	[ParameterHash]
	,	[GroupHash]
	,	[EventCount]
	,	[ItemCount]
	,	[RunDate])
	select
		[EntityID]
	,   [PartitionValue]
	,   [MessageID]
	,   [ParameterHash]
	,	[GroupHash]
	,   sum([EventCount])
	,	count(distinct [ItemId])
	,   @runDate
	from
		[mig_tracking].[RootItemEvent]
		where
			(@entitiesToProcess is null or [EntityID] in (select [EntityID] from @entities))
	group by
		[EntityID]
	,   [PartitionValue]
	,   [MessageID]
	,   [ParameterHash]
	,	[GroupHash]

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	-----------------------------------------------------
	-- PartitionEventCount
	-----------------------------------------------------
	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'PartitionEventCount, update', @logId output

	declare @addedCounts table(
		[MessageID] uniqueidentifier not null
	,   [EntityID] uniqueidentifier not null
	,   [PartitionValue] nvarchar(50) not null
	,   [InstanceCount] int not null
	,   [ItemCount] int not null
	)

	declare @existingCounts table(
		[MessageID] uniqueidentifier not null
	,   [EntityID] uniqueidentifier not null
	,   [PartitionValue] nvarchar(50) not null
	,   [InstanceCount] int not null
	,   [ItemCount] int not null
	)

	;with
		[latest] as (
			select
				[MessageID]
			,	[EntityID]
			,	[PartitionValue]
			,	[InstanceCount]
			,	[ItemCount]
			,	row_number() over (partition by [MessageID], [EntityID], [PartitionValue] order by [Time] desc) as [rn]
			from
				[mig_tracking].[PartitionEventCount]
		)
	insert @existingCounts
	select
		[MessageID]
	,	[EntityID]
	,	[PartitionValue]
	,	[InstanceCount]
	,	[ItemCount]
	from
		[latest]
	where
		[rn] = 1

	;with
		[newCounts] as (
			select
				[MessageID]
			,	[EntityID]
			,   [PartitionValue]
			,   sum([EventCount]) as [InstanceCount]
			,	count(distinct [ItemId]) as [ItemCount]
			from
				[mig_tracking].[RootItemEvent]
			where
				(@entitiesToProcess is null or [EntityID] in (select [EntityID] from @entities))
			group by
				[MessageID]
			,	[EntityID]
			,   [PartitionValue]
		)
	insert [mig_tracking].[PartitionEventCount] (
		[MessageID]
	,	[EntityID]
	,	[PartitionValue]
	,	[InstanceCount]
	,	[ItemCount]
	,   [Time]
	)
	output inserted.[MessageID], inserted.[EntityID], inserted.[PartitionValue], inserted.[InstanceCount], inserted.[ItemCount] into @addedCounts
	select
		coalesce(A.[MessageID], B.[MessageID])
	,   coalesce(A.[EntityID], B.[EntityID])
	,   coalesce(A.[PartitionValue], B.[PartitionValue])
	,   isnull(A.[InstanceCount], 0)
	,   isnull(A.[ItemCount], 0)
	,	@runDate
	from
		[newCounts] as A
		full outer join @existingCounts as B
		on
			A.[MessageID] = B.[MessageID]
		and A.[EntityID] = B.[EntityID]
		and A.[PartitionValue] = B.[PartitionValue]
		where
			(@entitiesToProcess is null or coalesce(A.[EntityID], B.[EntityID]) in (select [EntityID] from @entities))
		and (
				isnull(A.[InstanceCount], 0) <> isnull(B.[InstanceCount], 0)
			or	isnull(A.[ItemCount], 0) <> isnull(B.[ItemCount], 0)
		)

	-- Log count changes
	insert [mig_tracking].[Log]
	select
		A.[MessageID]
	,   A.[EntityID]
	,   A.[PartitionValue]
	,   case
			when B.[InstanceCount] is null then 0 -- Added
			else 1 -- Modified
		end
	,   @runDate
	,   @user
	,   'Instance Count (' + A.[PartitionValue] + ')'
	,   format(isnull(B.[InstanceCount], 0), 'N0')
	,   format(A.[InstanceCount], 'N0')
	from
		@addedCounts as A
		left outer join @existingCounts as B
		on
			A.[MessageID] = B.[MessageID]
		and	A.[EntityID] = B.[EntityID]
		and	A.[PartitionValue] = B.[PartitionValue]

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	-----------------------------------------------------
	-- Statistics
	-----------------------------------------------------
	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Statistics', @logId output;

	with
		export as (
			select *
			from
				[mig_director].[_Export_Statistics]()
			where
				[PartitionValue] is not null
		)
	,	import as (
			select *
			from [mig_director].[_Import_Statistics]()
			where
				[PartitionValue] is not null
		)
	,	delivery as (
			select *
			from [mig_director].[_Delivery_Statistics]()
			where
				[PartitionValue] is not null
		)
	,	newStatistic as (
			select
				A.[PartitionValue]
			,	A.[EntityID]
			,   A.[Count] - A.[Dropped] as [Available]
			,   A.[Discarded] as [ExpDiscarded]
			,   A.[Exported]
			,   isnull(B.[Discarded], 0) as [ImpDiscarded]
			,   isnull(B.[Imported], 0) as [Imported]
			,   isnull(C.[Discarded], 0) as [DelDiscarded]
			,   isnull(C.[Delivered], 0) as [Delivered]
			,   A.[Difference] + isnull(B.[Difference], 0) + isnull(C.[Difference], 0) as [Check]
			from
				export as A
				left outer join import as B
				on
					A.[PartitionValue] = B.[PartitionValue]
				and	A.[EntityID] = B.[EntityID]
				left outer join delivery as C
				on
					A.[PartitionValue] = C.[PartitionValue]
				and	A.[EntityID] = C.[EntityID]
			where
				(@entitiesToProcess is null or A.[EntityID] in (select [EntityID] from @entities))
		)
	,	curStatistic as (
			select *
			from
				[mig_tracking].[Statistic] as A
			where
				(@entitiesToProcess is null or A.[EntityID] in (select [EntityID] from @entities))
			and	A.[Time] = (
					select max([Time])
					from
						[mig_tracking].[Statistic]
					where
						[PartitionValue] = A.[PartitionValue]
					and	[EntityID] = A.[EntityID]
				)
		)
	insert [mig_tracking].[Statistic] (
		[PartitionValue]
	,	[EntityID]
	,	[Time]
	,	[Available]
	,	[ExpDiscarded]
	,	[Exported]
	,	[ImpDiscarded]
	,	[Imported]
	,	[DelDiscarded]
	,	[Delivered]
	)
	select
		coalesce(A.[PartitionValue], B.[PartitionValue]) as [PartitionValue]
	,	coalesce(A.[EntityID], B.[EntityID]) as [EntityID]
	,	@runDate as [RunDate]
	,   isnull(A.[Available], 0) as [Available]
	,   isnull(A.[ExpDiscarded], 0) as [ExpDiscarded]
	,   isnull(A.[Exported], 0) as [Exported]
	,   isnull(A.[ImpDiscarded], 0) as [ImpDiscarded]
	,   isnull(A.[Imported], 0) as [Imported]
	,   isnull(A.[DelDiscarded], 0) as [DelDiscarded]
	,   isnull(A.[Delivered], 0) as [Delivered]
	from
		newStatistic as A
		full outer join curStatistic as B
		on
			A.[PartitionValue] = B.[PartitionValue]
		and	A.[EntityID] = B.[EntityID]
	where
		A.[EntityID] is null -- Obsolete
	or	B.[EntityID] is null -- New
	or	B.[Available] <> A.[Available]
	or	B.[ExpDiscarded] <> A.[ExpDiscarded]
	or	B.[Exported] <> A.[Exported]
	or	B.[ImpDiscarded] <> A.[ImpDiscarded]
	or	B.[Imported] <> A.[Imported]
	or	B.[DelDiscarded] <> A.[DelDiscarded]
	or	B.[Delivered] <> A.[Delivered]

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	-----------------------------------------------------
	-- Parameters
	-----------------------------------------------------
	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 1, 'Parameters', @logId output

	truncate table [mig_tracking].[Parameter]

	insert [mig_tracking].[Parameter]
	select
		[ItemID]
	,   [ProjectType]
	,   [Name]
	,   [Value]
	,	[Description]
	from
		[mig_director].[RunParameter]
	where
		[Present] = 1
	and [Value] is not null

	exec [mig_director].[JobInstance_LogEntryUpdate] @logId, 0

	set @text =
'------------------------------------------------------------------
' + convert(varchar, getdate() - @start, 108) + ' Total
------------------------------------------------------------------'

	exec [mig_director].[JobInstance_LogEntryInsert] @jobInstanceId, 0, @text, @logId output

	-- Return the current non-zero partition counts for status updates in the project DB
	;with
		[latest] as (
			select
				[MessageID]
			,	[EntityID]
			,	[PartitionValue]
			,	[InstanceCount]
			,	row_number() over (partition by [MessageID], [EntityID], [PartitionValue] order by [Time] desc) as [rn]
			from
				[mig_tracking].[PartitionEventCount]
		)
	select
		[MessageID]
	,	[EntityID]
	,	[PartitionValue]
	from
		[latest]
	where
		[rn] = 1
	and	[InstanceCount] > 0
end
GO

PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- PartitionEventCount nonclustered index for ROW_NUMBER scans filtered by EntityID'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Creating index [IX_PartitionEventCount_EntityID] on [mig_tracking].[PartitionEventCount]...'
GO

IF NOT EXISTS (SELECT * FROM sys.indexes WHERE object_id = OBJECT_ID(N'[mig_tracking].[PartitionEventCount]') AND name = N'IX_PartitionEventCount_EntityID')
	CREATE NONCLUSTERED INDEX [IX_PartitionEventCount_EntityID]
		ON [mig_tracking].[PartitionEventCount] ([EntityID] ASC, [MessageID] ASC, [PartitionValue] ASC, [Time] DESC)
		INCLUDE ([InstanceCount], [ItemCount])
GO

PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- Prune obsolete messages: streaming source SPs for currently-occurring User and System events'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Creating procedure [mig_tracking].[EventUser_GetCurrent]...'
GO

CREATE OR ALTER procedure [mig_tracking].[EventUser_GetCurrent]
	@IncludeExport		bit
,	@IncludeImport		bit
,	@IncludeDelivery	bit
as
	set nocount on

	-----------------------------------------------------
	-- User events currently occurring in this track,
	-- consumed as a streamed TVP by
	-- [mig_tracking].[EventUser_KeepSet_Append] in the
	-- Project DB. Column order must match the receiving
	-- TVP type [EventUserKeyInput] one-for-one (ordinal
	-- mapping, no MessageID — that's resolved on the
	-- Project DB side via the MessageKey tuple).
	-----------------------------------------------------
	select distinct
		[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	[EventID]
	from
		[mig_valid].[EventUser]
	where
		(@IncludeExport		= 1 and [Step] = 0)
		or (@IncludeImport	= 1 and [Step] = 1)
		or (@IncludeDelivery	= 1 and [Step] = 2)
GO

PRINT N'Creating procedure [mig_tracking].[EventSystem_GetCurrent]...'
GO

CREATE OR ALTER procedure [mig_tracking].[EventSystem_GetCurrent]
	@IncludeExport		bit
,	@IncludeImport		bit
,	@IncludeDelivery	bit
as
	set nocount on

	-----------------------------------------------------
	-- System events currently occurring in this track,
	-- consumed as a streamed TVP by
	-- [mig_tracking].[EventSystem_KeepSet_Append] in the
	-- Project DB. Column order must match the receiving
	-- TVP type [EventSystemKeyInput] one-for-one (ordinal
	-- mapping). [Message] is VARCHAR(256) here; the TVP's
	-- [MessageFmt] is NVARCHAR(MAX) — DbDataReader -> TVP
	-- needs matching column types so convert + alias here.
	-----------------------------------------------------
	select distinct
		[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	convert(nvarchar(max), [Message]) as [MessageFmt]
	from
		[mig_valid].[EventSystem]
	where
		(@IncludeExport		= 1 and [Step] = 0)
		or (@IncludeImport	= 1 and [Step] = 1)
		or (@IncludeDelivery	= 1 and [Step] = 2)
GO

PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- Prune obsolete messages: PartitionEventCount cleanup TVP type and SPs'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Creating type [mig_tracking].[MessageIDInput]...'
GO

IF TYPE_ID(N'[mig_tracking].[MessageIDInput]') IS NULL
	CREATE TYPE [mig_tracking].[MessageIDInput] AS TABLE
	(
		[MessageID]		uniqueidentifier	not null
	)
GO

PRINT N'Creating procedure [mig_tracking].[PartitionEventCount_PruneIds_Append]...'
GO

CREATE OR ALTER procedure [mig_tracking].[PartitionEventCount_PruneIds_Append]
	@ids [mig_tracking].[MessageIDInput] readonly
as
	set nocount on

	-- Requires caller-owned temp: #PartitionPruneIds (created by DbTrack.PartitionPruneIds_Init).

	------------------------------------------------------------------------------------------------------
	-- Receives the project-wide #PruneDeleteSet stream as a TVP and inserts the MessageIDs into the
	-- caller-owned #PartitionPruneIds temp on the held DbTrack connection. The Prune Messages job
	-- creates #PartitionPruneIds with a primary key on MessageID; the not-exists predicate keeps the
	-- INSERT idempotent against TVP duplicates.
	------------------------------------------------------------------------------------------------------
	insert into #PartitionPruneIds ([MessageID])
	select i.[MessageID]
	from @ids as i
	where
		not exists (
			select 1
			from #PartitionPruneIds as k
			where k.[MessageID] = i.[MessageID]
		)
GO

PRINT N'Creating procedure [mig_tracking].[PartitionEventCount_PruneBatch]...'
GO

CREATE OR ALTER procedure [mig_tracking].[PartitionEventCount_PruneBatch]
	@BatchSize int
as
	set nocount on

	-- Requires caller-owned temp: #PartitionPruneIds (created by DbTrack.PartitionPruneIds_Init).

	------------------------------------------------------------------------------------------------------
	-- Deletes one batch of [PartitionEventCount] rows whose MessageID matches an entry in the
	-- caller-owned #PartitionPruneIds. All rows for a matched MessageID are eligible regardless of
	-- [Time] — the underlying message is gone, so every historical snapshot is unreachable. Manager
	-- calls this in a loop until rowcount = 0 so per-batch progress can be surfaced via
	-- IJobInstanceLogEntry.Update.
	------------------------------------------------------------------------------------------------------
	delete top (@BatchSize) from [mig_tracking].[PartitionEventCount]
	where [MessageID] in (select [MessageID] from #PartitionPruneIds)

	select @@rowcount as [Rows]
GO

PRINT N'Creating procedure [mig_tracking].[PartitionEventCount_RefreshStatistics]...'
GO

CREATE OR ALTER procedure [mig_tracking].[PartitionEventCount_RefreshStatistics]
as
	set nocount on

	------------------------------------------------------------------------------------------------------
	-- Refreshes optimizer statistics on [PartitionEventCount] after the Prune Messages job's
	-- per-track cleanup. The cleanup can shrink the table by orders of magnitude on projects with
	-- large historical data; explicit full-scan refresh keeps subsequent plans (Impact widget,
	-- Statistic list) calibrated for the post-delete cardinality without waiting for auto-stats.
	------------------------------------------------------------------------------------------------------
	update statistics [mig_tracking].[PartitionEventCount] with fullscan
GO
