Relational Queues

CREATE TABLE SyncQueue (
    EnqueuedAt _Timestamp,
    UpdatedAt _Timestamp,

    [Status] QueueState,
    [Type] _Type,
    [Step] QueueStep,
    AttemptNum AttemptNum,
    Response WebResponse DEFAULT '',
    Error WebResponse DEFAULT '',
    StartedAt _Timestamp,
    Duration Duration,
    ScheduledFor _Timestamp,

    CONSTRAINT PK_SyncQueue PRIMARY KEY (EnqueuedAt),

    CONSTRAINT SyncQueue_IsClassifiedBy_SyncQueueType
        FOREIGN KEY ([Type])
        REFERENCES SyncQueueType([Type]),

    CONSTRAINT SyncQueue_IsStatedBy_QueueStatus
        FOREIGN KEY ([Status])
        REFERENCES QueueStatus([Status])
);

CREATE TABLE QueueStatus (
    [Status] QueueState,

    CONSTRAINT PK_QueueStatus PRIMARY KEY CLUSTERED ([Status])
);

INSERT INTO QueueStatus([Status]) VALUES
    ('Pending'),
    ('Processing'),
    ('Error'),           -- Retryable
    ('Failed'),          -- Non-retryable (max attempts exceeded)
    ('Completed'),
    ('Cancelled'),
    ('Paused'),
    ('Terminated'),
    ('PartiallyCompleted');

CREATE OR ALTER FUNCTION QueueIsProcessable_fn (@Status QueueState)
RETURNS BIT AS BEGIN
    -- Can this item be picked up for processing?
    IF @Status IN ('Pending', 'Paused', 'Error')
        RETURN 1;
    RETURN 0;
END;

CREATE OR ALTER FUNCTION QueueIsFinished_fn (@Status QueueState)
RETURNS BIT AS BEGIN
    -- Is this item in a terminal state?
    IF @Status NOT IN ('Pending', 'Processing', 'Error', 'Paused')
        RETURN 1;
    RETURN 0;
END;

CREATE OR ALTER FUNCTION QueueIsEditable_fn (@Status QueueState)
RETURNS BIT AS BEGIN
    -- Can a worker write response/error data to this item?
    IF @Status NOT IN ('Cancelled', 'Paused', 'Terminated', 'Completed', 'PartiallyCompleted')
        RETURN 1;
    RETURN 0;
END;

CREATE OR ALTER PROCEDURE Next_Notification_trx
AS BEGIN

    DECLARE @EnqueuedAt _Timestamp;
    DECLARE @MaxAttempts _Int;
    DECLARE @ErrNo INT;
    DECLARE @RowCnt INT;

    IF (@@TRANCOUNT > 0) BEGIN
        RAISERROR(50012, 16, 1, 'Next_Notification_trx');
        GOTO EXIT_ERROR;
    END

    -- Read max attempts from centralized settings, default to 3
    SET @MaxAttempts = COALESCE(
        (SELECT ValInt FROM AppSettings WHERE Param = 'notification.maxAttempts'),
        3
    );

    BEGIN TRANSACTION Next_Notification_trx;

        -- READPAST skips rows locked by other workers
        -- UPDLOCK + ROWLOCK claims this specific row
        SELECT TOP (1)
            @EnqueuedAt = EnqueuedAt
        FROM Notification WITH (UPDLOCK, ROWLOCK, READPAST)
        WHERE
            [Status] IN ('Pending', 'Error')
            AND ScheduledFor <= SYSDATETIME()
            AND AttemptNum < @MaxAttempts
        ORDER BY
            ScheduledFor ASC;

        -- No work available — clean exit
        IF @EnqueuedAt IS NULL BEGIN
            COMMIT TRANSACTION Next_Notification_trx;
            RETURN 0;
        END

        -- Claim the item: set to Processing, bump attempt count
        UPDATE Notification
        SET
            [Status] = 'Processing',
            UpdatedAt = SYSDATETIME(),
            StartedAt = SYSDATETIME(),
            AttemptNum = AttemptNum + 1
        WHERE
            EnqueuedAt = @EnqueuedAt;

        SELECT @RowCnt = @@ROWCOUNT, @ErrNo = @@ERROR;

        IF (@ErrNo <> 0) GOTO EXIT_TRANSACTION;
        IF (@RowCnt <> 1) BEGIN
            RAISERROR(50005, 16, 1, 'Next_Notification_trx: Notification');
            GOTO EXIT_TRANSACTION;
        END

    COMMIT TRANSACTION Next_Notification_trx;

    -- Return the claimed item's key and concurrency token
    SELECT EnqueuedAt, UpdatedAt FROM Worker_Notification_V
    WHERE EnqueuedAt = @EnqueuedAt;

    RETURN 0;

EXIT_TRANSACTION:
    ROLLBACK TRANSACTION Next_Notification_trx;

EXIT_ERROR:
    RETURN 1;

END;

CREATE OR ALTER PROCEDURE Modify_Notification_trx
    @EnqueuedAt _Timestamp,
    @UpdatedAt _Timestamp,        -- optimistic concurrency token
    @Status QueueState,
    @Step QueueStep = NULL,        -- current step in multi-step jobs
    @Response WebResponse = NULL,
    @Error WebResponse = NULL,
    @ScheduledFor _Timestamp = NULL
AS BEGIN
    -- 1. Validate @@TRANCOUNT = 0
    -- 2. Classify the new and current status using QueueIs*_fn functions
    -- 3. Validate the transition is legal
    -- 4. Read max attempts from AppSettings (default to 3)
    -- 5. Enforce max attempts (force Failed if exceeded)
    -- 6. Calculate duration if finishing
    -- 7. UPDATE with WHERE UpdatedAt = @UpdatedAt (optimistic concurrency)
    --    Include Step = ISNULL(@Step, Step) to preserve or advance the step
    -- 8. Return updated item through worker view
END;

-- Base: shared queue columns
CREATE TABLE Notification (
    EnqueuedAt _Timestamp PRIMARY KEY,
    [Type] _Type,
    [Status] QueueState,
    [Step] QueueStep,
    AttemptNum AttemptNum,
    Response WebResponse DEFAULT '',
    Error WebResponse DEFAULT '',
    StartedAt _Timestamp,
    Duration Duration,
    ScheduledFor _Timestamp,
    UpdatedAt _Timestamp,

    CONSTRAINT Notification_IsDiscriminatedBy_NotificationType
        FOREIGN KEY ([Type]) REFERENCES NotificationType([Type]),
    CONSTRAINT Notification_IsStatedBy_QueueStatus
        FOREIGN KEY ([Status]) REFERENCES QueueStatus([Status])
);

-- Exclusive subtype: data specific to a password reset notification
CREATE TABLE Notification_ResetPassword (
    EnqueuedAt _Timestamp PRIMARY KEY,
    OTP OtpCode,
    UserID DbUserID,
    Username DbUsername,

    CONSTRAINT ResetPassword_Is_Notification
        FOREIGN KEY (EnqueuedAt) REFERENCES Notification(EnqueuedAt),
    CONSTRAINT ResetPassword_IsType_Notification
        CHECK (dbo.Notification_IsType_fn(EnqueuedAt, 'ResetPassword') = 1)
);

-- Inclusive subtypes: delivery channels (a notification can be both email AND sms)
CREATE TABLE Notification_Email (
    EnqueuedAt _Timestamp PRIMARY KEY,
    EmailTo Email,
    PreferredLanguage [Name] DEFAULT 'en',
    Response WebResponse DEFAULT '',
    Error WebResponse DEFAULT '',

    CONSTRAINT Email_Is_Notification
        FOREIGN KEY (EnqueuedAt) REFERENCES Notification(EnqueuedAt)
);