SqlPulse 2.0.1

SqlPulse

SqlPulse is an enterprise-grade, multi-targeted (.NET 8.0, .NET Standard 2.0, .NET Framework 4.6.2) real-time SQL Server table tracking engine built natively on Microsoft.Data.SqlClient.

It provides sub-millisecond data mutation event dispatching using low-overhead database mechanisms, guarded by a built-in internal Bounded Channel Backpressure Engine and high-performance kernel-level thread suspension (ManualResetEventSlim) to insulate calling applications from high-frequency transactional data spikes.

🏗️ Architectural Matrix & Topology

Before deploying this utility, enterprise architects must review the structural boundaries and select the appropriate watcher engine version based on the target table's write-velocity profiles:

⚖️ Architectural Trade-Offs & Scaling Boundaries (When to Pivot to Kafka)

SqlPulse is an optimal, zero-infrastructure solution for lightweight cache eviction and sequential microservice background execution workflows. However, it operates on a different engineering foundation than distributed log-based change data capture platforms like the Apache Kafka SQL Server Connector (Debezium).

Architects must evaluate the following scaling boundaries before determining the system design path:

1. Primary Engine Mechanisms

• SqlPulse: Utilizes native synchronous database transaction hooks (Triggers/Broker pipelines). This introduces a microsecond latency overhead directly into the user's write query loop (INSERT/UPDATE/DELETE).
• Kafka SQL Server Connector (Debezium): Utilizes out-of-band asynchronous log sniffing. It reads changes directly from the SQL Server transaction log (.ldf) using the native database CDC agent jobs after the user's query has already finished, ensuring absolute zero impact on primary query write speeds.

2. Backpressure & Failover Durability

• SqlPulse: Relies on an in-memory Bounded Channel buffer (System.Threading.Channels) inside the application container process context. If your application node crashes or stays down longer than your database retention configuration window, transient notification frames are lost.
• Kafka Connect Platform: Kafka acts as a massive, disk-backed sequential byte queue. If your downstream application microservice crashes for 24 hours, Kafka buffers millions of transaction logs safely on its own cluster disk nodes. The moment your service boots back up, it resumes reading from its exact byte offset without putting any stress on the relational database engine.

3. Horizontal Scale-Out Boundary

• SqlPulse: Lives directly inside the host process memory footprint of your application assembly. It is strictly optimized for single-node worker daemons or standalone integration workflows. If you scale out your web layer horizontally across multiple cloud container instances, multiple watchers will execute concurrent queries against the same SQL tables, creating catalog contention.
• Kafka Connect Cluster: Features native distributed state coordination (Consumer Groups). It splits high-volume table mutations across discrete topic partitions and distributes load evenly across many container worker instances without duplicate event processing.

📐 The Architect's Decision Matrix

• Choose SqlPulse if: Your transactional write velocity is low-to-medium, you want to protect your database storage from the massive disk inflation caused by full database CDC logging, and you want to avoid the high infrastructure cost of running a distributed Java-based Kafka cluster.
• Pivot to the Apache Kafka Connector if: You are working on ultra-high-velocity tables (thousands of concurrent inserts per second) where you cannot tolerate any extra latency on the write queries, you need multi-day buffer durability during application outages, or your service is designed to scale out aggressively horizontally across a large cloud cluster.

🔒 Security & DBA Provisioning Ledger

A developer or team leader cannot initialize this library without prior database adjustments. The following administrative permissions must be explicitly granted by the DBA to the application user account ({database_user}, if not 'administrator' or 'db_owner') inside the target database (replace {database_user} with actual database user):

📜 Idempotent Database Provisioning Script

-- Connect directly to your application database container (e.g., {YourDatabaseName})
PRINT 'STARTING CONGRUENCE CHECK FOR {database_user}';
DECLARE @PrincipalId INT = USER_ID('{database_user}');
DECLARE @TargetSchema NVARCHAR(128) = N'dbo';

-- 1. Database-Level Capabilities
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CONNECT' AND class = 0 AND state = 'G') GRANT CONNECT TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CONTROL' AND class = 0 AND state = 'G') GRANT CONTROL TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CREATE QUEUE' AND class = 0 AND state = 'G') GRANT CREATE QUEUE TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CREATE SERVICE' AND class = 0 AND state = 'G') GRANT CREATE SERVICE TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CREATE MESSAGE TYPE' AND class = 0 AND state = 'G') GRANT CREATE MESSAGE TYPE TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CREATE CONTRACT' AND class = 0 AND state = 'G') GRANT CREATE CONTRACT TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'CREATE PROCEDURE' AND class = 0 AND state = 'G') GRANT CREATE PROCEDURE TO [{database_user}];
IF NOT EXISTS (SELECT 1 FROM sys.database_permissions WHERE grantee_principal_id = @PrincipalId AND permission_name = 'SUBSCRIBE QUERY NOTIFICATIONS' AND class = 0 AND state = 'G') GRANT SUBSCRIBE QUERY NOTIFICATIONS TO [{database_user}];

-- 2. Schema-Level Functional Access (Required for data reads and execution channels)
EXEC('GRANT ALTER, CONTROL, SELECT, EXECUTE ON SCHEMA::[' + @TargetSchema + '] TO [{database_user}];');
PRINT '{database_user} COMPLIANCE CONFIGURATION SUCCESSFUL.';
GO

🚀 Application Integration Blueprint

The architecture is entirely decoupled from the global hosting framework. The calling application instantiates a watcher, binds an isolated handler method, and manages the execution lifecycle independently.

1. Mapping TrackNotification (Lightweight Event Loop)

/*
 Create a database and provide database user appropriate permissions as described in the README.md under section Security & DBA Provisioning Ledger.
 Create a table name it as Claims and add the following columns to it:
  - ClaimId (int, primary key, identity)
  - ClaimNumber (nvarchar(50))
  - ClaimantName (nvarchar(100))
  - DateOfLoss (datetime)
  - Status (nvarchar(20))
 Create some sample data in the Claims table to test the application.

 Add package references (or add through Nuget Package Manager)to the project for Microsoft.Extensions.Logging.Console (only for console application) and SqlPulse.
 */

using Microsoft.Extensions.Logging;
using SqlPulse.Contract;
using SqlPulse.EventArg;
using SqlPulse.Implementation;

bool keepRunning = true;

using var loggerFactory = LoggerFactory.Create(builder =>
{
    builder.AddConsole().SetMinimumLevel(LogLevel.Information);
});

string connectionString = "Server=DESKTOP-C21QPND;Database={YourDatabaseName};User Id={YourDBUser};Password={YourDBPassword};TrustServerCertificate=True;MultipleActiveResultSets=True;";

IWatchTracker watcher = new WatchTracker(loggerFactory);

using var watcherNotificationOnly = watcher.TrackNotification(new NotificationWatcherParameters(
        instanceName: "NotificationOnlyPipeline", // Create multiple instances of the watcher with different instance names to track same/multiple tables in parallel
        connectionString: connectionString,
        schemaName: "dbo",
        tableName: "Claims", 
        columnsToTrack: new[] { "ClaimId", "Status" }, 
        loggerFactory: loggerFactory, 
        whereClause: "status = 'Pending'", // Apply where clause to filter notifications based on specific conditions, make it null if you want to track all changes
        triggerOnStartup: false, // Natively executes a fast COUNT(1) to avoid empty startup iterations, make it true if you want to trigger a notification on startup
        maxSpikeQueueCapacity: 1000 // Thread safe buffer boundary limits
    ));

watcherNotificationOnly.TableChanged += OnDatabaseTableChanged;

try
{
    watcherNotificationOnly.Start();
    Console.ForegroundColor = ConsoleColor.Yellow;
    Console.WriteLine("\n[STATUS] Notification Only Pipeline fully initialized and actively monitoring table [dbo].[Claims]...");
    Console.WriteLine("[ACTION] Go ahead and open SSMS. Execute an INSERT, UPDATE, or DELETE statement on the table or simply change status column.");
    Console.ResetColor();
}
catch (Exception ex)
{
    Console.ForegroundColor = ConsoleColor.Red;
    Console.WriteLine($"\n!!! [CRITICAL LAUNCH ERROR] Mapped exceptions thrown during startup sequence: {ex.Message}");
    if (ex.InnerException != null)
    {
        Console.WriteLine($"Inner Details: {ex.InnerException.Message}");
    }
    Console.ResetColor();
}

bool showMenu = true;

while (keepRunning)
{
    if (showMenu)
    {
        Console.ForegroundColor = ConsoleColor.Cyan;
        Console.WriteLine("\n=== Notification Handler Event Control Menu ===");
        Console.WriteLine("1. Type 'P' to PAUSE the notification handler event.");
        Console.WriteLine("2. Type 'R' to RESUME the notification handler event.");
        Console.WriteLine("3. Type '0' to QUIT the application.");
        Console.ResetColor();
    }

    string input = Console.ReadLine();

    // 1. Try to convert the input into an integer
    if (int.TryParse(input, out int number))
    {
        if (number == 0)
        {
            watcherNotificationOnly.Dispose();
            keepRunning = false;
            Console.WriteLine("Exiting application... Goodbye!");
        }
        else
        {
            // 3. Process any other valid number here
            Console.WriteLine($"You entered: {number}. The square is {number * number}.");
        }
    }
    else
    {
        // 4. Handle invalid inputs (like letters or symbols)
        if (input.ToLower() == "p")
        {
            watcherNotificationOnly.Pause();
            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine("Notification handler event paused, now change status column. No notification should be tracked.");
            Console.WriteLine("Perform changes or INSERT/DELETE as many time as you want.");
            Console.ResetColor();
            showMenu = false;
        }
        else if (input.ToLower() == "r")
        {
            watcherNotificationOnly.Resume();
            //watcherNotificationOnly.Resume(true); //Use this overload to clear backpressure queue if you want to discard any pending notifications that were queued while paused.
            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine("Notification handler event resumed, now change status column. Notification should be tracked.");
            Console.ResetColor();
            showMenu = false;
        }
        else
        {
            Console.WriteLine("Invalid input. Please enter valid input.");
            showMenu = true;
        }
    }
}

void OnDatabaseTableChanged(object? sender, NotificationEventArgs e)
{
    Console.ForegroundColor = ConsoleColor.Yellow;
    Console.WriteLine($"\n>>> [LIGHTWEIGHT EVENT] Table [{e.SchemaName}].[{e.TableName}] encountered a raw {e.Action.ToString().ToUpper()} operation.");
    Console.WriteLine("    [INFO] This event is triggered without payload data for performance-sensitive scenarios.");
    Console.WriteLine(new string('-', 60));
    Console.ResetColor();
}

2. Mapping TrackPayload (Deep Data Capture Loop)

/*
 Create a database and provide database user appropriate permissions as described in the README.md under section Security & DBA Provisioning Ledger.
 Create a table name it as Claims and add the following columns to it:
  - ClaimId (int, primary key, identity)
  - ClaimNumber (nvarchar(50))
  - ClaimantName (nvarchar(100))
  - DateOfLoss (datetime)
  - Status (nvarchar(20))
 Create some sample data in the Claims table to test the application.

 Add package references (or add through Nuget Package Manager)to the project for Microsoft.Extensions.Logging.Console (only for console application) and SqlPulse.
 */

using Microsoft.Extensions.Logging;
using SqlPulse.Contract;
using SqlPulse.EventArg;
using SqlPulse.Implementation;

bool keepRunning = true;

using var loggerFactory = LoggerFactory.Create(builder =>
{
    builder.AddConsole().SetMinimumLevel(LogLevel.Information);
});

string connectionString = "Server=DESKTOP-C21QPND;Database={YourDatabaseName};User Id={YourDBUser};Password={YourDBPassword};TrustServerCertificate=True;MultipleActiveResultSets=True;";

IWatchTracker watcher = new WatchTracker(loggerFactory);

using var payloadWatcher = watcher.TrackPayload(new PayloadWatcherParameters(
        instanceName: "DeepClaimsWatcher",
        connectionString: connectionString,
        schemaName: "dbo",
        tableName: "Claims",
        loggerFactory: loggerFactory,
        maxSpikeQueueCapacity: 1000 // Thread safe buffer boundary limits
    ));

payloadWatcher.TableChanged += OnDatabaseMutationCaptured;

using var payloadWatcher1 = watcher.TrackPayload(new PayloadWatcherParameters(
        instanceName: "DeepClaimsWatcher1",
        connectionString: connectionString,
        schemaName: "dbo",
        tableName: "Claims",
        loggerFactory: loggerFactory,
        maxSpikeQueueCapacity: 1000 // Thread safe buffer boundary limits
    ));

payloadWatcher1.TableChanged += OnDatabaseMutationCaptured1;

try
{
    // 5. Build infrastructure elements and spin up the background Service Broker reader loop
    payloadWatcher.Start();
    payloadWatcher1.Start();
    Console.ForegroundColor = ConsoleColor.Green;
    Console.WriteLine("\n[STATUS] Pipeline fully initialized and actively monitoring table [dbo].[Claims]...");
    Console.WriteLine("[ACTION] Go ahead and open SSMS. Execute an INSERT, UPDATE, or DELETE statement on the table.");
    Console.WriteLine("Press [ENTER] at any time to exit the application context and drop tracking hooks.\n");
    Console.ResetColor();
}
catch (Exception ex)
{
    Console.ForegroundColor = ConsoleColor.Red;
    Console.WriteLine($"\n!!! [CRITICAL LAUNCH ERROR] Mapped exceptions thrown during startup sequence: {ex.Message}");
    if (ex.InnerException != null)
    {
        Console.WriteLine($"Inner Details: {ex.InnerException.Message}");
    }
    Console.ResetColor();
}

/// <summary>
/// Isolated handler method executed on background ThreadPool threads whenever a row mutation occurs.
/// </summary>
void OnDatabaseMutationCaptured(object? sender, PayloadChangedEventArgs e)
{
    Console.ForegroundColor = ConsoleColor.Green;
    Console.WriteLine($"\n>>> [EVENT RECEIVED] Mutation captured in [{e.SchemaName}].[{e.TableName}]! Operation: {e.Action.ToString().ToUpper()}");

    switch (e.Action)
    {
        case WatchAction.Insert:
            if (e.AfterRecord != null)
            {
                Console.WriteLine("    [INSERT DATA] Added fresh row values:");
                foreach (var kvp in e.AfterRecord)
                {
                    Console.WriteLine($"      - {kvp.Key}: {kvp.Value}");
                }
            }
            break;

        case WatchAction.Update:
            if (e.BeforeRecord != null && e.AfterRecord != null)
            {
                Console.WriteLine("    [UPDATE DATA] Column variances detected:");
                foreach (var key in e.AfterRecord.Keys)
                {
                    var oldVal = e.BeforeRecord.ContainsKey(key) ? e.BeforeRecord[key] : "NULL";
                    var newVal = e.AfterRecord[key];

                    if (oldVal?.ToString() != newVal?.ToString())
                    {
                        Console.WriteLine($"      - Field [{key}]: Changed from '{oldVal}' ---> '{newVal}'");
                    }
                }
            }
            break;

        case WatchAction.Delete:
            if (e.BeforeRecord != null)
            {
                Console.WriteLine("    [DELETE DATA] Historical snapshot of removed row data:");
                foreach (var kvp in e.BeforeRecord)
                {
                    Console.WriteLine($"      - {kvp.Key}: {kvp.Value}");
                }
            }
            break;
    }
    Console.WriteLine(new string('-', 60));
    Console.ResetColor();
}

/// <summary>
/// Isolated handler method executed on background ThreadPool threads whenever a row mutation occurs.
/// </summary>
void OnDatabaseMutationCaptured1(object? sender, PayloadChangedEventArgs e)
{
    Console.ForegroundColor = ConsoleColor.Green;
    Console.WriteLine($"\n>>> [EVENT RECEIVED] Mutation captured in [{e.SchemaName}].[{e.TableName}]! Operation: {e.Action.ToString().ToUpper()}");

    switch (e.Action)
    {
        case WatchAction.Insert:
            if (e.AfterRecord != null)
            {
                Console.WriteLine("    [INSERT DATA] Added fresh row values:");
                foreach (var kvp in e.AfterRecord)
                {
                    Console.WriteLine($"      - {kvp.Key}: {kvp.Value}");
                }
            }
            break;

        case WatchAction.Update:
            if (e.BeforeRecord != null && e.AfterRecord != null)
            {
                Console.WriteLine("    [UPDATE DATA] Column variances detected:");
                foreach (var key in e.AfterRecord.Keys)
                {
                    var oldVal = e.BeforeRecord.ContainsKey(key) ? e.BeforeRecord[key] : "NULL";
                    var newVal = e.AfterRecord[key];

                    if (oldVal?.ToString() != newVal?.ToString())
                    {
                        Console.WriteLine($"      - Field [{key}]: Changed from '{oldVal}' ---> '{newVal}'");
                    }
                }
            }
            break;

        case WatchAction.Delete:
            if (e.BeforeRecord != null)
            {
                Console.WriteLine("    [DELETE DATA] Historical snapshot of removed row data:");
                foreach (var kvp in e.BeforeRecord)
                {
                    Console.WriteLine($"      - {kvp.Key}: {kvp.Value}");
                }
            }
            break;
    }
    Console.WriteLine(new string('-', 60));
    Console.ResetColor();
}

bool showMenu = true;

while (keepRunning)
{
    if (showMenu)
    {
        Console.ForegroundColor = ConsoleColor.Cyan;
        Console.WriteLine("\n=== Notification Handler Event Control Menu ===");
        Console.WriteLine("1. Type 'P' to PAUSE the notification handler event.");
        Console.WriteLine("2. Type 'R' to RESUME the notification handler event.");
        Console.WriteLine("3. Type '0' to QUIT the application.");
        Console.ResetColor();
    }

    string input = Console.ReadLine();

    // 1. Try to convert the input into an integer
    if (int.TryParse(input, out int number))
    {
        if (number == 0)
        {
            payloadWatcher.Dispose(); // Ensure proper cleanup of resources before exiting
            keepRunning = false;
            Console.WriteLine("Exiting application... Goodbye!");
        }
        else
        {
            // 3. Process any other valid number here
            Console.WriteLine($"You entered: {number}. The square is {number * number}.");
        }
    }
    else
    {
        // 4. Handle invalid inputs (like letters or symbols)
        if (input.ToLower() == "p")
        {
            payloadWatcher.Pause();
            payloadWatcher1.Pause();
            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine("Notification handler event paused, now change status column. No notification should be tracked.");
            Console.WriteLine("Perform changes or INSERT/DELETE as many time as you want.");
            Console.ResetColor();
            showMenu = false;
        }
        else if (input.ToLower() == "r")
        {
            payloadWatcher.Resume();
            payloadWatcher1.Resume();
            //payloadWatcher.Resume(true); //Use this overload to clear backpressure queue if you want to discard any pending notifications that were queued while paused.
            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine("Notification handler event resumed, now change status column. Notification should be tracked.");
            Console.ResetColor();
            showMenu = false;
        }
        else
        {
            Console.WriteLine("Invalid input. Please enter valid input.");
            showMenu = true;
        }
    }
}

⚡ Spike Controls: Pause and Resume Operations

Both tracking engines feature high-performance Pause() and Resume() execution structures. When an application calls .Pause(), database message streams continue dropping into the bounded memory channel at maximum hardware speed. However, the background worker thread halts at the kernel layer using ManualResetEventSlim, completely protecting your downstream app code blocks from load thrashing.

public void ManageSystemPeakOverload(IWatcher<NotificationEventArgs> engine)
{
    // 1. Suspend application handler dispatching during API spikes
    engine.Pause(); 
    
    // --> Incoming database modifications safely queue up inside RAM cache...
    
    // 2. Resume execution loops. The thread instantly wakes up and drains the cache sequentially
    engine.Resume(); 
}

public void ManageSystemPeakOverload(IWatcher<PayloadChangedEventArgs> engine)
{
    // 1. Suspend application handler dispatching during API spikes
    engine.Pause(); 
    
    // --> Incoming database modifications safely queue up inside RAM cache...
    
    // 2. Resume execution loops. The thread instantly wakes up and drains the cache sequentially
    engine.Resume(); 
}