<commit_before>----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[AddItem]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[AddItem]
GO
CREATE PROCEDURE [Queue].[AddItem]
	@Key nvarchar(250),
	@Value varbinary(max)
AS
BEGIN
	SET NOCOUNT ON;

    INSERT INTO Queue.SubscriptionStorage ([Key],Value) VALUES (@Key,@Value)
    SELECT SCOPE_IDENTITY();
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[CreateQueueIfMissing]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[CreateQueueIfMissing]
GO
CREATE PROCEDURE [Queue].[CreateQueueIfMissing]
	@Queue nvarchar(50),
	@Endpoint nvarchar(250)
AS
BEGIN
	SET NOCOUNT ON;
	DECLARE @QueueId int
	
			SELECT @QueueId = QueueId
			FROM Queue.Queues
			WHERE QueueName = @Queue
			AND Endpoint=@Endpoint;
			
			if (@QueueId is null)
				BEGIN
					INSERT INTO Queue.Queues (QueueName,Endpoint) VALUES (@Queue,@Endpoint)
					SELECT @QueueId = SCOPE_IDENTITY()
				END

		SELECT @QueueId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[EnqueueMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[EnqueueMessage]
GO
CREATE PROCEDURE [Queue].[EnqueueMessage]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@Payload varbinary(MAX),
	@Headers nvarchar(2000),
	@ProcessingUntil datetime,
	@CreatedAt datetime,
	@ExpiresAt datetime
AS
BEGIN
	SET NOCOUNT ON;

    DECLARE @QueueId int;
    
    EXEC Queue.GetAndAddQueue @Endpoint,@Queue,@Subqueue,@QueueId=@QueueId OUTPUT;
		
	INSERT INTO Queue.Messages (QueueId,Payload,ProcessingUntil,ExpiresAt,Processed,Headers,CreatedAt) VALUES (@QueueId,@Payload,ISNULL(@ProcessingUntil,GetDate()),@ExpiresAt,0,@Headers,@CreatedAt)
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[ExtendMessageLease]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[ExtendMessageLease]
GO
CREATE PROCEDURE [Queue].[ExtendMessageLease]
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

	UPDATE Queue.Messages
	SET ProcessingUntil = DateAdd(mi,10,GetDate())
	WHERE MessageId=@MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[GetAndAddQueue]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[GetAndAddQueue]
GO
CREATE PROCEDURE [Queue].[GetAndAddQueue]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@QueueId int OUTPUT
AS
BEGIN
	SET NOCOUNT ON;

	SELECT @QueueId = QueueId FROM Queue.Queues WHERE QueueName = @Queue AND Endpoint = @Endpoint;
	if (@QueueId is null)
		BEGIN
			INSERT INTO Queue.Queues (QueueName,Endpoint) VALUES (@Queue,@Endpoint)
			SELECT @QueueId = SCOPE_IDENTITY()
		END
		
	IF (@Subqueue is not null)
		BEGIN
			DECLARE @SubqueueId AS int
			
			SELECT @SubqueueId = s.QueueID FROM Queue.Queues p INNER JOIN Queue.Queues s ON p.QueueId = s.ParentQueueId WHERE p.QueueName = @Queue AND p.Endpoint = @Endpoint AND s.QueueName = @Subqueue
			if (@SubqueueId is null)
				BEGIN
					INSERT INTO Queue.Queues (QueueName,ParentQueueId,Endpoint) VALUES (@Subqueue,@QueueId,@Endpoint)
					SELECT @SubqueueId = SCOPE_IDENTITY()
				END
				
			SET @QueueId = @SubqueueId
		END

	SET NOCOUNT OFF;
	RETURN @QueueId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[GetItemsByKey]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[GetItemsByKey]
GO
CREATE PROCEDURE [Queue].[GetItemsByKey]
	@Key nvarchar(250)
AS
BEGIN
	SET NOCOUNT ON;

    SELECT *
    FROM Queue.SubscriptionStorage
    WHERE ([Key]=@Key)
    
    DELETE FROM Queue.SubscriptionStorage
    WHERE ([Key]=@Key)
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[MarkMessageAsReady]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[MarkMessageAsReady]
GO
CREATE PROCEDURE [Queue].[MarkMessageAsReady]
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

--Debug
	/*
	UPDATE Queue.Messages
	SET Processed = 1
	WHERE MessageId = @MessageId
	*/
	DELETE Queue.Messages
	WHERE MessageId = @MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[MoveMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[MoveMessage]
GO
CREATE PROCEDURE [Queue].[MoveMessage]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

    DECLARE @QueueId int;
        
    EXEC Queue.GetAndAddQueue @Endpoint,@Queue,@Subqueue,@QueueId=@QueueId OUTPUT;

	UPDATE Queue.Messages
	SET QueueId = @QueueId
	WHERE MessageId=@MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[PeekMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[PeekMessage]
GO
CREATE PROCEDURE [Queue].[PeekMessage]
	@QueueId int
AS
BEGIN
	SET NOCOUNT ON;
	
	SELECT TOP 1 *
	FROM Queue.Messages
	WHERE isnull(ExpiresAt,DATEADD(mi,1,GetDate())) > GetDate()
	AND Processed=0
	AND ProcessingUntil<GetDate()
	AND QueueId = @QueueId<commit_msg>Updated procedures for better handling of timeout messages.<commit_after>----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[AddItem]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[AddItem]
GO
CREATE PROCEDURE [Queue].[AddItem]
	@Key nvarchar(250),
	@Value varbinary(max)
AS
BEGIN
	SET NOCOUNT ON;

    INSERT INTO Queue.SubscriptionStorage ([Key],Value) VALUES (@Key,@Value)
    SELECT SCOPE_IDENTITY();
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[CreateQueueIfMissing]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[CreateQueueIfMissing]
GO
CREATE PROCEDURE [Queue].[CreateQueueIfMissing]
	@Queue nvarchar(50),
	@Endpoint nvarchar(250)
AS
BEGIN
	SET NOCOUNT ON;
	DECLARE @QueueId int
	
			SELECT @QueueId = QueueId
			FROM Queue.Queues
			WHERE QueueName = @Queue
			AND Endpoint=@Endpoint;
			
			if (@QueueId is null)
				BEGIN
					INSERT INTO Queue.Queues (QueueName,Endpoint) VALUES (@Queue,@Endpoint)
					SELECT @QueueId = SCOPE_IDENTITY()
				END

		SELECT @QueueId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[EnqueueMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[EnqueueMessage]
GO
CREATE PROCEDURE [Queue].[EnqueueMessage]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@Payload varbinary(MAX),
	@Headers nvarchar(2000),
	@ProcessingUntil datetime,
	@CreatedAt datetime,
	@ExpiresAt datetime
AS
BEGIN
	SET NOCOUNT ON;

    DECLARE @QueueId int;
    
    EXEC Queue.GetAndAddQueue @Endpoint,@Queue,@Subqueue,@QueueId=@QueueId OUTPUT;
		
	INSERT INTO Queue.Messages (QueueId,Payload,ProcessingUntil,ExpiresAt,Processed,Headers,CreatedAt) VALUES (@QueueId,@Payload,ISNULL(@ProcessingUntil,GetDate()),@ExpiresAt,0,@Headers,@CreatedAt)
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[ExtendMessageLease]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[ExtendMessageLease]
GO
CREATE PROCEDURE [Queue].[ExtendMessageLease]
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

	UPDATE Queue.Messages
	SET ProcessingUntil = DateAdd(mi,10,GetDate())
	WHERE MessageId=@MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[GetAndAddQueue]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[GetAndAddQueue]
GO
CREATE PROCEDURE [Queue].[GetAndAddQueue]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@QueueId int OUTPUT
AS
BEGIN
	SET NOCOUNT ON;

	SELECT @QueueId = QueueId FROM Queue.Queues WHERE QueueName = @Queue AND Endpoint = @Endpoint AND ParentQueueId IS NULL;
	if (@QueueId is null)
		BEGIN
			INSERT INTO Queue.Queues (QueueName,Endpoint) VALUES (@Queue,@Endpoint)
			SELECT @QueueId = SCOPE_IDENTITY()
		END
		
	IF (@Subqueue is not null)
		BEGIN
			DECLARE @SubqueueId AS int
			
			SELECT @SubqueueId = s.QueueID FROM Queue.Queues p INNER JOIN Queue.Queues s ON p.QueueId = s.ParentQueueId WHERE p.QueueName = @Queue AND p.Endpoint = @Endpoint AND s.QueueName = @Subqueue
			if (@SubqueueId is null)
				BEGIN
					INSERT INTO Queue.Queues (QueueName,ParentQueueId,Endpoint) VALUES (@Subqueue,@QueueId,@Endpoint)
					SELECT @SubqueueId = SCOPE_IDENTITY()
				END
				
			SET @QueueId = @SubqueueId
		END

	SET NOCOUNT OFF;
	RETURN @QueueId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[GetItemsByKey]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[GetItemsByKey]
GO
CREATE PROCEDURE [Queue].[GetItemsByKey]
	@Key nvarchar(250)
AS
BEGIN
	SET NOCOUNT ON;

    SELECT *
    FROM Queue.SubscriptionStorage
    WHERE ([Key]=@Key)
    
    DELETE FROM Queue.SubscriptionStorage
    WHERE ([Key]=@Key)
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[MarkMessageAsReady]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[MarkMessageAsReady]
GO
CREATE PROCEDURE [Queue].[MarkMessageAsReady]
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

--Debug
	/*
	UPDATE Queue.Messages
	SET Processed = 1
	WHERE MessageId = @MessageId
	*/
	DELETE Queue.Messages
	WHERE MessageId = @MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[MoveMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[MoveMessage]
GO
CREATE PROCEDURE [Queue].[MoveMessage]
	@Endpoint nvarchar(250),
	@Queue nvarchar(50),
	@Subqueue nvarchar(50),
	@MessageId int
AS
BEGIN
	SET NOCOUNT ON;

    DECLARE @QueueId int;
        
    EXEC Queue.GetAndAddQueue @Endpoint,@Queue,@Subqueue,@QueueId=@QueueId OUTPUT;

	UPDATE Queue.Messages
	SET QueueId = @QueueId,
	ProcessingUntil = GetDate()
	WHERE MessageId=@MessageId
END
GO
----
IF  EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[Queue].[PeekMessage]') AND type in (N'P', N'PC'))
DROP PROCEDURE [Queue].[PeekMessage]
GO
CREATE PROCEDURE [Queue].[PeekMessage]
	@QueueId int
AS
BEGIN
	SET NOCOUNT ON;
	
	SELECT TOP 1 *
	FROM Queue.Messages
	WHERE isnull(ExpiresAt,DATEADD(mi,1,GetDate())) > GetDate()
	AND Processed=0
	AND ProcessingUntil<GetDate()
	AND QueueId = @QueueId
{'fraction_non_alphanumeric': 0.12342395587076438, 'fraction_numerical': 0.00847123719464145, 'mean_word_length': 4.732919254658385, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '10840', 'n_tokens_mistral': 3969, 'n_tokens_neox': 3568, 'n_words': 980}