Lenni's Technology Blog

{
  "EventHub": {
    "HostName": "ces-namespace.servicebus.windows.net",
    "Name": "ces-hub",
    "SasToken": "paste-your-sas-token-here"
  },
  "BlobStorage": {
    "ConnectionString": "paste-your-blob-connection-string-here",
    "ContainerName": "ces-checkpoint"
  }
}

using Azure;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Processor;
using Azure.Messaging.EventHubs.Consumer;
using Azure.Storage.Blobs;
using Microsoft.Extensions.Configuration;
using System;
using System.Collections.Generic;
using System.IO;
using System.Text.Json;
using System.Threading.Tasks;

namespace CESClient
{
  public class Program
  {
    private static int _eventCount;

    // Add methods here

  }
}

public static async Task Main(string[] args)
{
  // Say hello
  Console.WriteLine("SQL Server 2025 Change Event Streaming Client");
  Console.WriteLine();
  Console.Write("Initializing... ");

  // Load configuration from appsettings.json
  var config = new ConfigurationBuilder()
      .SetBasePath(Directory.GetCurrentDirectory())
      .AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
      .Build();

  // Create a blob container client that the event processor will use for checkpointing
  var blobStorageConnectionString = config["BlobStorage:ConnectionString"];
  var blobStorageContainerName = config["BlobStorage:ContainerName"];

  var storageClient = new BlobContainerClient(blobStorageConnectionString, blobStorageContainerName);

  // Create an event processor client to process events in the event hub
  var eventHubHostName = config["EventHub:HostName"];
  var eventHubName = config["EventHub:Name"];
  var sasToken = config["EventHub:SasToken"];

  var processor = new EventProcessorClient(
      storageClient,                                        // checkpoint store
      EventHubConsumerClient.DefaultConsumerGroupName,      // Basic tier: one consumer group (e.g., $Default)
      eventHubHostName,
      eventHubName,
      new AzureSasCredential(sasToken)
  );

  // Register handlers for processing events and errors
  processor.ProcessEventAsync += ProcessEventHandler;
  processor.ProcessErrorAsync += ProcessErrorHandler;

  // Start listening for events
  Console.Write("starting... ");
  _eventCount = 0;

  await processor.StartProcessingAsync();

  Console.WriteLine("waiting... press any key to stop.");
  Console.ReadKey(intercept: true);

  // Stop listening for events
  await processor.StopProcessingAsync();

  Console.WriteLine("Stopped");
}

private static async Task ProcessEventHandler(ProcessEventArgs eventArgs)
{
  try
  {
    // Deserialize the event data
    using var doc = JsonDocument.Parse(eventArgs.Data.Body.ToArray());
    var root = doc.RootElement;
    var dataJson = root.GetProperty("data");

    using var innerDoc = JsonDocument.Parse(dataJson.GetString());
    var data = innerDoc.RootElement;

    Console.WriteLine($"Processing event... #{++_eventCount}");

    // Deserialize the "current" and "old" fields in the eventrow property of the event data to dictionaries
    var operation = root.GetProperty("operation").GetString();
    var cols = data.GetProperty("eventsource").GetProperty("cols").EnumerateArray();
    var current = JsonSerializer.Deserialize<Dictionary<string, string>>(data.GetProperty("eventrow").GetProperty("current").GetString());
    var old = JsonSerializer.Deserialize<Dictionary<string, string>>(data.GetProperty("eventrow").GetProperty("old").GetString());

    DisplayEventMetadata(eventArgs, root, data);

    switch (operation)
    {
      case "INS":
        ProcessInsert(cols, current);
        break;
      case "UPD":
        ProcessUpdate(cols, current, old);
        break;
      case "DEL":
        ProcessDelete(cols, old);
        break;
    }

    Console.WriteLine();
    Console.WriteLine(new string('-', 80));
    Console.WriteLine();

    // Persist progress so we don't reprocess this event on restart
    await eventArgs.UpdateCheckpointAsync();
  }
  catch (Exception ex)
  {
    Console.ForegroundColor = ConsoleColor.Red;
    Console.WriteLine(ex.Message);
    Console.ResetColor();
  }
}

private static void DisplayEventMetadata(ProcessEventArgs eventArgs, JsonElement root, JsonElement data)
{
  Console.WriteLine("Event Args");
  Console.WriteLine($"  Sequence:Offset => {eventArgs.Data.SequenceNumber}:{eventArgs.Data.Offset}");
  Console.WriteLine();
  Console.WriteLine("Event Data");
  Console.WriteLine($"  Spec version:       {root.GetProperty("specversion").GetString()}");
  Console.WriteLine($"  Operation:          {root.GetProperty("type").GetString()}");
  Console.WriteLine($"  Time:               {root.GetProperty("time").GetString()}");
  Console.WriteLine($"  Event ID:           {root.GetProperty("id").GetString()}");
  Console.WriteLine($"  Logical ID:         {root.GetProperty("logicalid").GetString()}");
  Console.WriteLine($"  Operation:          {root.GetProperty("operation").GetString()}");
  Console.WriteLine($"  Data content type:  {root.GetProperty("datacontenttype").GetString()}");
  Console.WriteLine();
  Console.WriteLine("Data");
  Console.WriteLine($"  Database:           {data.GetProperty("eventsource").GetProperty("db").GetString()}");
  Console.WriteLine($"  Schema:             {data.GetProperty("eventsource").GetProperty("schema").GetString()}");
  Console.WriteLine($"  Table:              {data.GetProperty("eventsource").GetProperty("tbl").GetString()}");
  Console.WriteLine();
}

private static void ProcessInsert(JsonElement.ArrayEnumerator cols, Dictionary<string, string> current)
{
  Console.WriteLine("Operation: Insert");
  Console.ForegroundColor = ConsoleColor.Green;

  foreach (var col in cols)
  {
    var name = col.GetProperty("name").GetString();
    Console.WriteLine($"\t{name}: {current[name]}");
  }

  Console.ResetColor();
}

private static void ProcessUpdate(JsonElement.ArrayEnumerator cols, Dictionary<string, string> current, Dictionary<string, string> old)
{
  Console.WriteLine("Operation: Update");

  foreach (var col in cols)
  {
    var name = col.GetProperty("name").GetString();

    if (old.Count > 0 && current[name] != old[name])
    {
      Console.ForegroundColor = ConsoleColor.Yellow;
      Console.WriteLine($"\t{name}: {current[name]} (old: {old[name]})");
      Console.ResetColor();
    }
    else
    {
      Console.WriteLine($"\t{name}: {current[name]}");
    }
  }
}

private static void ProcessDelete(JsonElement.ArrayEnumerator cols, Dictionary<string, string> old)
{
  Console.WriteLine("Operation: Delete");
  Console.ForegroundColor = ConsoleColor.Red;

  foreach (var col in cols)
  {
    var name = col.GetProperty("name").GetString();
    Console.WriteLine($"\t{name}: {old[name]}");
  }

  Console.ResetColor();
}

private static Task ProcessErrorHandler(ProcessErrorEventArgs e)
{
  Console.ForegroundColor = ConsoleColor.Red;
  Console.WriteLine(e.Exception.Message);
  Console.ResetColor();
  return Task.CompletedTask;
}

EXEC CreateOrder @CustomerId = 1

EXEC CreateOrderDetail @OrderId = 1, @ProductId = 1, @Quantity = 2
EXEC CreateOrderDetail @OrderId = 1, @ProductId = 2, @Quantity = 1

EXEC DeleteOrder @OrderId = 1

UPDATE Customer SET City = 'Chicago' WHERE CustomerId = 1

UPDATE Product SET UnitPrice = UnitPrice * 0.8 WHERE Category = 'Camera'

UPDATE TableWithNoPK SET ItemName = 'Stove' WHERE Id = 3

# Install the general Azure cmdlets (this can take up to 20 minutes)
Install-Module -Name Az -Scope CurrentUser -Repository PSGallery -Force

# Install the Event Hub module (runs quickly)
Install-Module -Name Az.EventHub -Scope CurrentUser -Force

function Generate-SasToken {

    # Provide values for the following resources:
    $resourceGroupName  = "ces-demo-rg"
    $namespaceName      = "ces-namespace"
    $eventHubName       = "ces-hub"
    $policyName         = "ces-policy"

    # Login to Azure and select the Azure Subscription
    Connect-AzAccount -InformationAction SilentlyContinue | Out-Null

    # Validate the existence of the specified resource group, event hub namespace, and event hub
    Get-AzResourceGroup -Name $resourceGroupName -ErrorAction Stop | Out-Null
    Get-AzEventHubNamespace -ResourceGroupName $resourceGroupName -Name $namespaceName -ErrorAction Stop | Out-Null
    Get-AzEventHub -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -Name $eventHubName -ErrorAction Stop | Out-Null

    # Get the event hub authorization policy (it must have Manage rights)
    $policy = Get-AzEventHubAuthorizationRule -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -EventHubName $eventHubName -AuthorizationRuleName $policyName -ErrorAction SilentlyContinue

    if (-not ("Manage" -in $policy.Rights)) {
        throw "Authorization rule '$policyName' does not exist, or is missing the required 'Manage' right"
    }

    # Get the Primary Key of the Shared Access Policy
    $keys = Get-AzEventHubKey -ResourceGroupName $resourceGroupName -NamespaceName $namespaceName -EventHubName $eventHubName -AuthorizationRuleName $policyName

    if (-not $keys) {
        throw "Could not obtain Azure Event Hub Key"
    }

    if (-not $keys.PrimaryKey) {
        throw "Could not obtain Primary Key"
    }

    $primaryKey = ($keys.PrimaryKey) 

    # Define a function to create the SAS token
    function Create-SasToken {
        param ([string]$resourceUri, [string]$keyName, [string]$key)

        $sinceEpoch = [datetime]::UtcNow - [datetime]"1970-01-01"
        $expiry = [int]$sinceEpoch.TotalSeconds + (60 * 60 * 24 * 31 * 6)  # 6 months
        $stringToSign = [System.Web.HttpUtility]::UrlEncode($resourceUri) + "`n" + $expiry
        $hmac = New-Object System.Security.Cryptography.HMACSHA256
        $hmac.Key = [Text.Encoding]::UTF8.GetBytes($key)
        $signature = [Convert]::ToBase64String($hmac.ComputeHash([Text.Encoding]::UTF8.GetBytes($stringToSign)))
        $sasToken = "SharedAccessSignature sr=$([System.Web.HttpUtility]::UrlEncode($resourceUri))&sig=$([System.Web.HttpUtility]::UrlEncode($signature))&se=$expiry&skn=$keyName"

        return $sasToken
    }

    # Construct the resource URI for the SAS token
    $resourceUri = "https://$namespaceName.servicebus.windows.net/$eventHubName"

    # Generate the SAS token using the primary key from the new policy
    $sasToken = Create-SasToken -resourceUri $resourceUri -keyName $policyName -key $primaryKey

    # Output the SAS token
    Write-Host "`n-- Generated SAS Token --" -ForegroundColor Gray
    Write-Host $sasToken -ForegroundColor White
    Write-Host "-- End of generated SAS Token --`n" -ForegroundColor Gray

    # Copy the SAS token to the clipboard
    $sasToken | Set-Clipboard
    Write-Host "The generated SAS token has been copied to the clipboard." -ForegroundColor Green
}

Generate-SasToken

Set-ExecutionPolicy -Scope Process -ExecutionPolicy Bypass

.\Generate-SasToken.ps1

-- Generated SAS Token --
<your SAS token>
-- End of generated SAS Token --

-- Create the demo database
USE master
GO

CREATE DATABASE CesDemo
GO

USE CesDemo
GO

-- Create some demo tables
CREATE TABLE Customer (
  CustomerId    int IDENTITY PRIMARY KEY,
  CustomerName  varchar(50),
  City          varchar(20)
)
GO

SET IDENTITY_INSERT Customer ON
INSERT INTO Customer
  (CustomerId,  CustomerName,               City) VALUES
  (1,           'Shutter Bros Wholesale',   'New York'),
  (2,           'Aperture Supply Co.',      'Los Angeles')
SET IDENTITY_INSERT Customer OFF

CREATE TABLE Product (
  ProductId     int IDENTITY PRIMARY KEY,
  Name          varchar(80),
  Color         varchar(15),
  Category      varchar(20),
  UnitPrice     decimal(8, 2),
  ItemsInStock  smallint
)
GO

SET IDENTITY_INSERT Product ON
INSERT INTO Product
  (ProductId, Name,                                  Color,     Category,       UnitPrice,  ItemsInStock) VALUES 
  (1,         'Canon EOS R5 Mirrorless Camera',      'Black',   'Camera',       3899.99,    10),
  (2,         'Nikon Z6 II Mirrorless Camera',       'Silver',  'Camera',       1996.95,    8),
  (3,         'Sony NP-FZ100 Rechargeable Battery',  'Black',   'Accessory',    78.00,      25)
SET IDENTITY_INSERT Product OFF

CREATE TABLE [Order] (
  OrderId       int IDENTITY PRIMARY KEY,
  CustomerId    int REFERENCES Customer(CustomerId),
  OrderDate     datetime2
)
GO

CREATE TABLE OrderDetail (
  OrderDetailId int IDENTITY PRIMARY KEY,
  OrderId       int REFERENCES [Order](OrderId),
  ProductId     int REFERENCES Product(ProductId),
  Quantity      smallint
)
GO

-- This table lacks Primary Key. Combining that with IncludeAllColumns = 0 results in events that
-- have no primary key, which is essentially useless
CREATE TABLE TableWithNoPK (
  Id        int IDENTITY,
  ItemName  varchar(50)
)
GO

INSERT INTO TableWithNoPK (ItemName) VALUES
  ('Camera'),
  ('Automobile'),
  ('Oven'),
  ('Couch')
GO

-- Create a DML trigger on OrderDetail that updates the ItemsInStock column in the Product table
-- based on the Quantity column in the OrderDetail table
CREATE TRIGGER trgUpdateItemsInStock ON OrderDetail AFTER INSERT, UPDATE, DELETE
AS
BEGIN
  -- Handle insert
  IF EXISTS (SELECT * FROM inserted) AND NOT EXISTS (SELECT * FROM deleted)
    UPDATE Product
    SET ItemsInStock = p.ItemsInStock - i.Quantity
    FROM
      Product AS p
      INNER JOIN inserted AS i ON p.ProductId = i.ProductId

  -- Handle update
  ELSE IF EXISTS (SELECT * FROM inserted) AND EXISTS (SELECT * FROM deleted) AND UPDATE(Quantity)
    UPDATE Product
    SET ItemsInStock = p.ItemsInStock + d.Quantity - i.Quantity
    FROM
      Product AS p
      INNER JOIN inserted AS i ON p.ProductId = i.ProductId
      INNER JOIN deleted AS d ON p.ProductId = d.ProductId

  -- Handle delete
  ELSE IF EXISTS (SELECT * FROM deleted) AND NOT EXISTS (SELECT * FROM inserted)
    UPDATE Product
    SET ItemsInStock = p.ItemsInStock + d.Quantity
    FROM
      Product AS p
      INNER JOIN deleted AS d ON p.ProductId = d.ProductId
END
GO

-- Add some procs to handle orders
CREATE OR ALTER PROC CreateOrder
  @CustomerId int
AS
BEGIN
  INSERT INTO [Order](CustomerId, OrderDate)
  VALUES (@CustomerId, SYSDATETIME())

  SELECT OrderId = SCOPE_IDENTITY()
END
GO

CREATE OR ALTER PROC CreateOrderDetail
  @OrderId int,
  @ProductId int,
  @Quantity smallint
AS
BEGIN
  INSERT INTO OrderDetail (OrderId, ProductId, Quantity)
  VALUES (@OrderId, @ProductId, @Quantity)

  SELECT OrderDetailId = SCOPE_IDENTITY()
END
GO

CREATE OR ALTER PROC DeleteOrder
  @OrderId int
AS
BEGIN
  BEGIN TRANSACTION
    DELETE FROM OrderDetail WHERE OrderId = @OrderId
    DELETE FROM [Order] WHERE OrderId = @OrderId
  COMMIT TRANSACTION
END
GO

CREATE MASTER KEY ENCRYPTION BY PASSWORD = 'H@rd2Gue$$P@$$w0rd'

CREATE DATABASE SCOPED CREDENTIAL SqlCesCredential
WITH
  IDENTITY = 'SHARED ACCESS SIGNATURE',
  SECRET = '<your SAS token>'

EXEC sys.sp_enable_event_stream

SELECT * FROM sys.databases WHERE is_event_stream_enabled = 1

EXEC sys.sp_create_event_stream_group
  @stream_group_name      = 'SqlCesGroup',
  @destination_location   = 'ces-namespace.servicebus.windows.net/ces-hub',
  @destination_credential = SqlCesCredential,
  @destination_type       = 'AzureEventHubsAmqp'

-- Customer: full row in each event, no old values
EXEC sys.sp_add_object_to_event_stream_group
  @stream_group_name = 'SqlCesGroup',
  @object_name = 'dbo.Customer',
  @include_old_values = 0,      -- do not include old values on updates/deletes
  @include_all_columns = 1      -- include all columns even if unchanged

-- Product: only changed columns, include old values (important for inventory diffs)
EXEC sys.sp_add_object_to_event_stream_group
  @stream_group_name = 'SqlCesGroup',
  @object_name = 'dbo.Product',
  @include_old_values = 1,      -- include old values for changed columns
  @include_all_columns = 0      -- only include changed columns

-- Order: only changed columns, include old values (for auditing changes)
EXEC sys.sp_add_object_to_event_stream_group
  @stream_group_name = 'SqlCesGroup',
  @object_name = 'dbo.Order',
  @include_old_values = 1,      -- include old values for changed columns
  @include_all_columns = 0      -- only include changed columns

-- OrderDetail: only changed columns, include old values (quantity updates matter)
EXEC sys.sp_add_object_to_event_stream_group
  @stream_group_name = 'SqlCesGroup',
  @object_name = 'dbo.OrderDetail',
  @include_old_values = 1,      -- include old values for changed columns
  @include_all_columns = 0      -- only include changed columns

-- TableWithNoPK: demonstrates CES limitations without a primary key
EXEC sys.sp_add_object_to_event_stream_group
  @stream_group_name = 'SqlCesGroup',
  @object_name = 'dbo.TableWithNoPK',
  @include_old_values = 0,      -- no old values
  @include_all_columns = 0      -- changed columns only (essentially useless without a PK)

EXEC sp_help_change_feed_table @source_schema = 'dbo', @source_name = 'Customer'
EXEC sp_help_change_feed_table @source_schema = 'dbo', @source_name = 'Product'
EXEC sp_help_change_feed_table @source_schema = 'dbo', @source_name = 'Order'
EXEC sp_help_change_feed_table @source_schema = 'dbo', @source_name = 'OrderDetail'
EXEC sp_help_change_feed_table @source_schema = 'dbo', @source_name = 'TableWithNoPK'

{
	"specversion": "1.0",
	"type": "com.microsoft.SQL.CES.DML.V1",
	"source": "\/",
	"id": "cc3fcdca-09c0-4f46-a8d3-5d0c3c1eb85a",
	"logicalid": "8376457a-17af-49f4-b9ea-0d5071f515f4:0000002C000007300011:00000000000000000002",
	"time": "2025-06-30T12:29:46.290Z",
	"datacontenttype": "application\/avro-json",
	"operation": "UPD",
	"segmentindex": 1,
	"finalsegment": true,
	"data": "{\n  \"eventsource\": {\n    \"db\": \"CesDemo\",\n    \"schema\": \"dbo\",\n    \"tbl\": \"Product\",\n    \"cols\": [\n      {\n        \"name\": \"ProductId\",\n        \"type\": \"int\",\n        \"index\": 0\n      },\n      {\n        \"name\": \"ItemsInStock\",\n        \"type\": \"smallint\",\n        \"index\": 5\n      }\n    ],\n    \"pkkey\": [\n      {\n        \"columnname\": \"ProductId\",\n        \"value\": \"2\"\n      }\n    ],\n    \"transaction\": {\n      \"commitlsn\": \"0000002C:00000730:0011\",\n      \"beginlsn\": \"0000002C:00000730:000C\",\n      \"sequencenumber\": 2,\n      \"committime\": \"2025-06-30T12:29:46.290Z\"\n    }\n  },\n  \"eventrow\": {\n    \"old\": \"{\\\"ProductId\\\": \\\"2\\\", \\\"ItemsInStock\\\": \\\"8\\\"}\",\n    \"current\": \"{\\\"ProductId\\\": \\\"2\\\", \\\"ItemsInStock\\\": \\\"7\\\"}\"\n  }\n}"
}

{
	"eventsource": {
		"db": "CesDemo",
		"schema": "dbo",
		"tbl": "Product",
		"cols": [
			{
				"name": "ProductId",
				"type": "int",
				"index": 0
			},
			{
				"name": "ItemsInStock",
				"type": "smallint",
				"index": 5
			}
		],
		"pkkey": [
			{
				"columnname": "ProductId",
				"value": "2"
			}
		],
		"transaction": {
			"commitlsn": "0000002C:00000730:0011",
			"beginlsn": "0000002C:00000730:000C",
			"sequencenumber": 2,
			"committime": "2025-06-30T12:29:46.290Z"
		}
	},
	"eventrow": {
		"old": "{\"ProductId\": \"2\", \"ItemsInStock\": \"8\"}",
		"current": "{\"ProductId\": \"2\", \"ItemsInStock\": \"7\"}"
	}
}