PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- EventSystem streaming bulk upsert, bridge-back, and UpdateStates streaming'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

PRINT N'Dropping obsolete procedure [mig_tracking].[EventSystem_InsertOrUpdate]...'
DROP PROCEDURE IF EXISTS [mig_tracking].[EventSystem_InsertOrUpdate]
GO

PRINT N'Dropping obsolete procedure [mig_tracking].[Event_List]...'
DROP PROCEDURE IF EXISTS [mig_tracking].[Event_List]
GO

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

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

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

CREATE OR ALTER PROCEDURE [mig_tracking].[EventSystem_InsertIfMissing]
	@events		[mig_tracking].[EventSystemInput] READONLY
,	@runDate	datetime
as
	set nocount on

	------------------------------------------------------------------------------------------------------
	-- Snapshot the existing 'S' rows into a temp table keyed on the dedup tuple plus a SHA-1 hash of
	-- MessageFmt. The bulk insert below anti-joins @events against this snapshot using the hash AND the
	-- raw MessageFmt — the hash narrows the candidate set so the optimizer picks a hash match anti-join,
	-- and the string compare confirms equality (defending against hash collisions). Going directly
	-- against [mig_tracking].[EventMessage] forces a per-row WHERE NOT EXISTS that key-looks-up
	-- MessageFmt for every probe — fine when the table holds a handful of distinct system events,
	-- prohibitive when a customer feedback file lands tens or hundreds of thousands of unique messages.
	--
	-- isnull(MessageFmt, '') applied on both sides so the comparison stays byte-identical regardless
	-- of nullability differences between the source column and the TVP.
	------------------------------------------------------------------------------------------------------
	drop table if exists #existing

	select
		[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	isnull([MessageFmt], N'') as [MessageFmt]
	,	hashbytes('SHA1', isnull([MessageFmt], N'')) as [MessageFmtHash]
	into #existing
	from [mig_tracking].[EventMessage]
	where [EventType] = 'S'

	------------------------------------------------------------------------------------------------------
	-- IMPORTANT: The 'where not exists' below determines the values on the event that constitutes the key
	-- for a MessageID. This is used in the Portal to map from the EventSystem in the Migration DB to a
	-- the MessageID in the Project DB. If these keys are changed in this Stored Procedure, it is vital
	-- to reflect the same changes in MigFx.Migration.Api.Infrastructure\Domain\MessageKey.cs
	------------------------------------------------------------------------------------------------------
	insert [mig_tracking].[EventMessage] (
		[MessageID]
	,	[EventType]
	,	[Engine]
	,	[Step]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	[MessageFmt]
	,	[RunDate]
	)
	select
		newid()
	,	'S'
	,	src.[Engine]
	,	src.[Step]
	,	src.[Disposition]
	,	src.[Impact]
	,	src.[Receiver]
	,	src.[MessageFmt]
	,	@runDate
	from
		@events as src
	where
		not exists (
			select 1
			from #existing as e
			where
				e.[Engine]			= src.[Engine]
			and	e.[Step]			= src.[Step]
			and e.[Disposition]		= src.[Disposition]
			and e.[Impact]			= src.[Impact]
			and e.[Receiver]		= src.[Receiver]
			and e.[MessageFmtHash]	= hashbytes('SHA1', isnull(src.[MessageFmt], N''))
			and e.[MessageFmt]		= isnull(src.[MessageFmt], N'')
		)
GO

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

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

	-----------------------------------------------------
	-- System Events with assigned MessageIDs (consumed
	-- as a streamed TVP by [mig_tracking].[Actual_Update]
	-- in the Migration DB; see DbTrack.Update on the
	-- consumer side)
	-----------------------------------------------------
	select
		[MessageID]
	,	[Step]
	,	[Engine]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	[MessageFmt]
	from [mig_tracking].[EventMessage]
	where [EventType] = 'S'
GO

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

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

	-----------------------------------------------------
	-- User Events with assigned MessageIDs (consumed
	-- as a streamed TVP by [mig_tracking].[Actual_Update]
	-- in the Migration DB; see DbTrack.Update on the
	-- consumer side)
	-----------------------------------------------------
	select
		[MessageID]
	,	[Step]
	,	[Engine]
	,	[Disposition]
	,	[Impact]
	,	[Receiver]
	,	[EventID]
	from [mig_tracking].[EventMessage]
	where [EventType] = 'U'
GO

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

IF TYPE_ID(N'[mig_tracking].[OccurringEventInput]') IS NULL
	CREATE TYPE [mig_tracking].[OccurringEventInput] AS TABLE
	(
		[MessageID]		uniqueidentifier	not null
	,	[EntityID]		uniqueidentifier	not null
	,	[PartitionValue]	nvarchar(50)		not null
	)
GO

PRINT N'Altering procedure [mig_tracking].[UpdateStates] (TVP parameter)...'
GO

CREATE OR ALTER procedure [mig_tracking].[UpdateStates]
	@occurring [mig_tracking].[OccurringEventInput] readonly
,	@user varchar(256)
,   @trackID uniqueidentifier
,	@runDate datetime2
as
	declare @updatedStates table(
		[MessageID] uniqueidentifier not null
	,	[EntityID] uniqueidentifier not null
	,	[PartitionValue] nvarchar(50) null
	)

	-- Initial EntityEventAnchor to register WorkLevel and TeamAnchor for new events
	;with
		msg as (
			select distinct
			[MessageID]
		,	[EntityID]
			from 
				@occurring
		)
	,	anchor as (
			select 
				B.[EntityID]
			,	A.[TeamAnchor]
			from
				[mig_project].[BusinessEntity] as A
				inner join [mig_project].[Entity] as B
				on
					A.[BusinessEntityID] = B.[BusinessEntityID]
		)
	insert [mig_tracking].[EntityEventAnchor] (
		[MessageID]
	,	[EntityID]
	,	[WorkLevel]
	,	[TeamAnchor]
	)
	select 
		A.[MessageID]
	,	A.[EntityID]
	,	case C.[Receiver]
			when 1 then 1 -- 1:Migration -> 1:Team
			else 0 -- 0:Source and 2:Target -> 0:Delegate
		end
	,	B.[TeamAnchor]
	from
		msg as A
		inner join anchor as B
		on
			A.[EntityID] = B.[EntityID]
		inner join [mig_tracking].[EventMessage] as C
		on
			A.[MessageID] = C.[MessageID]
		left outer join [mig_tracking].[EntityEventAnchor] as D
		on
			A.[MessageID] = D.[MessageID]
		and A.[EntityID] = D.[EntityID]
	where
		D.[MessageID] is null -- Only if the EntityEventAnchor does not exist
	and ( 
			-- Only add rows if necessary (not default values)
			B.[TeamAnchor] is not null
		or	C.[Receiver] in (0, 2) -- 0:Source, 2:Target
	)

	-- PartitionEventState
	;with
		[latest] as (
			select
				[MessageID]
			,	[EntityID]
			,	[PartitionValue]
			,	[UserState]
			,	row_number() over (partition by [MessageID], [EntityID], [PartitionValue] order by [Time] desc) as [rn]
			from
				[mig_tracking].[PartitionEventState]
			where
				[TrackID] = @trackID
		)
	,	[resolved] as (
			select
				[MessageID]
			,	[EntityID]
			,	[PartitionValue]
			,	[UserState]
			from
				[latest]
			where
				[rn] = 1
			and	[UserState] = 2 -- Resolved
		)
	insert [mig_tracking].[PartitionEventState]
	output inserted.[MessageID], inserted.[EntityID], inserted.[PartitionValue] into @updatedStates
	select
		@trackID
	,	A.[MessageID]
	,	A.[EntityID]
	,	A.[PartitionValue]
	,	@runDate
	,	1 -- Regression
	from
		[resolved] as A
		inner join @occurring as B
		on
			A.[MessageID] = B.[MessageID]
		and	A.[EntityID] = B.[EntityID]
		and	A.[PartitionValue] = B.[PartitionValue]

	-- EntityEventState
	;with
		[entities] as (
			select
				[MessageID] 
			,	[EntityID] 
			from @occurring
			group by
				[MessageID] 
			,	[EntityID] 
		)
	,	[delegated] as (
			select
				[MessageID] 
			,	[EntityID] 
			from	
				[mig_tracking].[EntityEventAnchor]
			where
				[WorkLevel] = 0 -- Delegate (inverted and left outer join, because the default (no EntityEventAnchor row) is WorkLevel.Team)
		)
	,	[latest] as (
			select
				[MessageID]
			,	[EntityID]
			,	[UserState]
			,	row_number() over (partition by [MessageID], [EntityID] order by [Time] desc) as [rn]
			from
				[mig_tracking].[EntityEventState]
			where
				[TrackID] = @trackID
		)
	,	[resolved] as (
			select
				A.[MessageID]
			,	A.[EntityID]
			,	A.[UserState]
			from
				[latest] as A
				left outer join [delegated] as B
				on
					A.[MessageID] = B.[MessageID]
				and	A.[EntityID] = B.[EntityID]
			where
				A.[rn] = 1
			and B.[MessageID] is null
			and A.[UserState] = 2 -- Resolved
		)
	insert [mig_tracking].[EntityEventState]
	output inserted.[MessageID], inserted.[EntityID], null into @updatedStates
	select
		@trackID
	,	A.[MessageID]
	,	A.[EntityID]
	,	@runDate
	,	1 -- Regression
	from
		[resolved] as A
		inner join [entities] as B
		on
			A.[MessageID] = B.[MessageID]
		and	A.[EntityID] = B.[EntityID]

	-- Log the state changes
	insert [mig_tracking].[Log] (
		[TrackID]
	,	[MessageID]
	,	[EntityID]
	,	[PartitionValue]
	,	[LogType]
	,	[LogTime]
	,	[LogUser]
	,	[Property]
	,	[Before]
	,	[After]
	)
	select
		@trackID
	,	[MessageID]
	,   [EntityID]
	,   [PartitionValue]
	,   1 -- Modified
	,   @runDate
	,   @user
	,   case
			when [PartitionValue] is null then 'Team State'
			else 'User State'
		end
	,   'Resolved'
	,   'Regression'
	from
		@updatedStates
GO

PRINT N'----------------------------------------------------------------------------------------------------------------------'
PRINT N'-- Prune obsolete messages: TVP key types and ProjectDB SPs'
PRINT N'----------------------------------------------------------------------------------------------------------------------'
GO

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

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

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

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

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

CREATE OR ALTER procedure [mig_tracking].[EventUser_KeepSet_Append]
	@keys [mig_tracking].[EventUserKeyInput] readonly
as
	set nocount on

	-- Requires caller-owned temp: #PruneKeepSet (created by DbProject.PruneStaging_Init).

	------------------------------------------------------------------------------------------------------
	-- Resolve User-event identification keys to MessageIDs by joining @keys against [EventMessage] on the
	-- key tuple defined in MessageKey.cs. Caller (.NET) owns #PruneKeepSet — created on the held
	-- DbProject connection before the first SP call, dropped on connection close.
	-- This SP is invoked once per in-use track during the streaming forward leg; existing MessageIDs are
	-- skipped via the not-exists predicate so multi-track unions are idempotent.
	------------------------------------------------------------------------------------------------------
	insert into #PruneKeepSet ([MessageID])
	select distinct em.[MessageID]
	from
		@keys as k
		inner join [mig_tracking].[EventMessage] as em
			on em.[EventType]		= 'U'
			and em.[Engine]			= k.[Engine]
			and em.[Step]			= k.[Step]
			and em.[Disposition]	= k.[Disposition]
			and em.[Impact]			= k.[Impact]
			and em.[Receiver]		= k.[Receiver]
			and em.[EventID]		= k.[EventID]
	where
		not exists (
			select 1
			from #PruneKeepSet as ks
			where ks.[MessageID] = em.[MessageID]
		)
GO

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

CREATE OR ALTER procedure [mig_tracking].[EventSystem_KeepSet_Append]
	@keys [mig_tracking].[EventSystemKeyInput] readonly
as
	set nocount on

	-- Requires caller-owned temps: #PruneKeepSet, #ExistingSystemMessage (created by DbProject.PruneStaging_Init).

	------------------------------------------------------------------------------------------------------
	-- Resolve System-event identification keys to MessageIDs by joining @keys against the snapshot
	-- #ExistingSystemMessage that was built once per job by PruneStaging_Init. The snapshot carries
	-- pre-computed MessageFmtHash values to keep the JOIN narrow on table sizes where MessageFmt is
	-- nvarchar(max) (not directly indexable). Mirrors the SHA-1 narrow + raw-string-confirm pattern
	-- from [EventSystem_InsertIfMissing] — the matching contract MUST stay in sync with that SP and
	-- with MessageKey.cs.
	------------------------------------------------------------------------------------------------------
	insert into #PruneKeepSet ([MessageID])
	select distinct e.[MessageID]
	from
		@keys as k
		inner join #ExistingSystemMessage as e
			on e.[Engine]			= k.[Engine]
			and e.[Step]			= k.[Step]
			and e.[Disposition]		= k.[Disposition]
			and e.[Impact]			= k.[Impact]
			and e.[Receiver]		= k.[Receiver]
			and e.[MessageFmtHash]	= hashbytes('SHA1', isnull(k.[MessageFmt], N''))
			and e.[MessageFmt]		= isnull(k.[MessageFmt], N'')
	where
		not exists (
			select 1
			from #PruneKeepSet as ks
			where ks.[MessageID] = e.[MessageID]
		)
GO

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

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

	-- Requires caller-owned temps: #PruneKeepSet, #PruneDeleteSet (created by DbProject.PruneStaging_Init).

	------------------------------------------------------------------------------------------------------
	-- Materialise the prune candidate set into #PruneDeleteSet (caller-owned temp on the held DbProject
	-- connection). A row is a candidate iff:
	--   - its EventType is in the user-checked set (User / System), AND
	--   - its Step is in the user-checked set (Export / Import / Delivery), AND
	--   - its MessageID is NOT in #PruneKeepSet (i.e., not currently occurring on any in-use track), AND
	--   - it has no [Comment] row (preserve user-annotated events), AND
	--   - it does not appear on either side of a [DerivedMessage] row (preserve derivation lineage,
	--     child or parent).
	-- Returns the candidate count via a single-row result set so the caller can log the workload size
	-- before entering the batch loop.
	------------------------------------------------------------------------------------------------------
	insert into #PruneDeleteSet ([MessageID])
	select em.[MessageID]
	from [mig_tracking].[EventMessage] as em
	where
		(
			(@IncludeUserEvents = 1 and em.[EventType] = 'U')
			or (@IncludeSystemEvents = 1 and em.[EventType] = 'S')
		)
		and (
			(@IncludeExport = 1 and em.[Step] = 0)
			or (@IncludeImport = 1 and em.[Step] = 1)
			or (@IncludeDelivery = 1 and em.[Step] = 2)
		)
		and not exists (
			select 1
			from #PruneKeepSet as ks
			where ks.[MessageID] = em.[MessageID]
		)
		and not exists (
			select 1
			from [mig_tracking].[Comment] as c
			where c.[MessageID] = em.[MessageID]
		)
		and not exists (
			select 1
			from [mig_tracking].[DerivedMessage] as dm
			where dm.[MessageID] = em.[MessageID]
				or dm.[ParentMessageID] = em.[MessageID]
		)

	select count_big(*) as [Count] from #PruneDeleteSet
GO

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

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

	-- Requires caller-owned temp: #PruneDeleteSet (created by DbProject.PruneStaging_Init).

	------------------------------------------------------------------------------------------------------
	-- Drain up to @BatchSize MessageIDs from #PruneDeleteSet, then DELETE the matching rows from each
	-- satellite table that references MessageID, finishing with [EventMessage] itself. Returns the count
	-- of MessageIDs processed so the caller can loop until 0.
	-- Cascade scope (no FKs in DB — explicit chain):
	--   Log, PartitionEventUser, PartitionEventState, EntityEventTag, EntityEventState, EntityEventAnchor,
	--   EventMessage. Comment and DerivedMessage are intentionally excluded — their presence prevented
	--   the candidate from entering #PruneDeleteSet in the first place.
	-- The drain + cascade run inside an explicit transaction with set xact_abort on so that any runtime
	-- error (deadlock victim, timeout, lock escalation) rolls the whole batch back as a unit. The drain
	-- DELETE on #PruneDeleteSet is rolled back too, so the next batch call sees the same MessageIDs and
	-- retries them — no partial-cascade orphans, no lost work.
	------------------------------------------------------------------------------------------------------
	begin transaction

	declare @batch table ([MessageID] uniqueidentifier primary key)
	declare @rows int

	delete top (@BatchSize) from #PruneDeleteSet
	output deleted.[MessageID] into @batch

	set @rows = @@rowcount

	if 0 = @rows
	begin
		commit transaction
		select 0 as [Rows]
		return
	end

	delete from [mig_tracking].[Log]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[PartitionEventUser]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[PartitionEventState]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[EntityEventTag]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[EntityEventState]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[EntityEventAnchor]
	where [MessageID] in (select [MessageID] from @batch)

	delete from [mig_tracking].[EventMessage]
	where [MessageID] in (select [MessageID] from @batch)

	commit transaction

	select @rows as [Rows]
GO

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

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

	------------------------------------------------------------------------------------------------------
	-- Refreshes optimizer statistics on the seven tables the prune cascade touches: Log,
	-- PartitionEventUser, PartitionEventState, EntityEventTag, EntityEventState, EntityEventAnchor,
	-- EventMessage. Called by the Prune Messages job after the batch loop drains so that subsequent
	-- queries (Impact widget, Statistic list) get plans sized for the post-delete row counts. Without
	-- this, a large prune leaves stale stats and downstream queries can pick plans calibrated for the
	-- pre-delete cardinality, which times out on tables that just shrank by orders of magnitude.
	------------------------------------------------------------------------------------------------------
	update statistics [mig_tracking].[Log] with fullscan
	update statistics [mig_tracking].[PartitionEventUser] with fullscan
	update statistics [mig_tracking].[PartitionEventState] with fullscan
	update statistics [mig_tracking].[EntityEventTag] with fullscan
	update statistics [mig_tracking].[EntityEventState] with fullscan
	update statistics [mig_tracking].[EntityEventAnchor] with fullscan
	update statistics [mig_tracking].[EventMessage] with fullscan
GO
