DEV Community: Mario Meyrelles

Deploy an Azure Function App connected to Cosmos DB with secrets on Key Vault using Bicep

Mario Meyrelles — Fri, 28 Jan 2022 17:57:58 +0000

Introduction

If you are using only Azure, you can work using Bicep, which is a recent IaC tool designed to help in the deployment of resources on Azure. You can use Terraform, Pulumi, Arm Templates or even, Azure CLI / Powershell to manipulate azure. Since there are many introductory posts on Bicep, I will go directly to the point and show how to deploy a potentially usable application on Azure. This work required hours of trial-and-error, googling and troubleshooting. It has been a somewhat painful journey, yet, rewarding journey. I hope you find this valuable for your work.

In this sample, I demonstrate how to structure your scripts to create the resources, store secrets on Azure Key Vault and I will show how to connect those secrets to the Function app. This example also covers a more involved scenario, using Azure RBAC for accessing Azure Key Vault as well as the creation of custom roles to configure the permissions needed.

The source code is available here: Source Code

What we are deploying?

In the image below, we can see the resources that have been created after running this deployment:

Fig 1: Resources created and configured on this resource group.

We also deploy an RBAC-only Azure Key Vault instance and save some keys inside it:

Fig 2: Azure Key Vault configured, with secrets deployed

Then, we create a Cosmos DB account in Serverless mode. We configure it with a Database and a Collection. Its access keys are stored on Key Vault during the deployment:

Fig 3: Cosmos DB Configured - Account, Database and two Containers.

Now we are ready to create the Function app. It depends on some components that we also create - Log Analytics, App Insights, Storage Account and an App Service Plan configured as Consumption Plan. As we can see in the pictures below, the configurations are linked directly to the Azure Key Vault and are exposed as common environment variables.

Fig 4: Function App deployed with success.

Fig 5: Function App is running - this indicates that the function has been able to be loaded correctly.

Fig 6: Some settings are connected directly to the Key Vault. There is a visual indication that the value has been loaded with success.

To make this work we create a User-Assigned Managed Identity and attach it to the Function App. With this enabled, the Function App can access the secrets stored in Key Vault using Azure RBAC. We also give access to the user who is deploying the application to manipulate.

Fig 7: User-Assigned Managed Identity.

Fig 8: System-Assigned Managed Identity is also necessary.

Please note that in this sample we don't configure VNet integration, IP restrictions, security groups. I don't have these requirements for my project. Now let's dive into the details of this script. There is a lot of see here :)

Structure of the script

The script is composed of many scripts that are connected on main.bicep:

@description('Application Name - change it!')
param appName string = 'sampleApp'

param location string = resourceGroup().location
param tenantId string = tenant().tenantId

@description('This is the object id of the user who will do the deployment on Azure. Can be your user id on AAD. Discover it running [az ad signed-in-user show] and get the [objectId] property.')
param deploymentOperatorId string 

// a 4-char suffix to add to the various names of azure resources to help them be unique, but still, predictable
var appSuffix = substring(uniqueString(resourceGroup().id),0,4)

// grants access to the operator of this deployment with specific roles like key vault access.
module operatorSetup 'operatorSetup.bicep' = {
  name: 'operatorSetup-deployment'
  params: {
    operatorPrincipalId: deploymentOperatorId
    appName: appName
  }
}

// creates an user-assigned managed identity that will used by different azure resources to access each other.
module msi 'msi.bicep' = {
  name: 'msi-deployment'
  params: {
    location: location
    managedIdentityName: '${appName}Identity'
    operatorRoleDefinitionId: operatorSetup.outputs.roleId
  }
}

// creates a key vault in this resource group
module keyvault 'keyvault.bicep' = {
  name: 'keyvault-deployment'
  params: {
    location: location
    appName: appName
    tenantId: tenantId
  }
}

// creates the cosmos db account and database with some containers configured. Saves connection string in keyvault.
module cosmos 'cosmosDb.bicep' = {
  name: 'cosmos-deployment'
  params:{
    cosmosAccountId: '${appName}-${appSuffix}'
    location: location
    cosmosDbName: appName
    keyVaultName: keyvault.outputs.keyVaultName
  }
}

// creates a Log Analytics + Application Insights instance
module logAnalytics 'logAnalytics.bicep' = {
  name: 'log-analytics-deployment'
  params: {
    appName: appName
  }
}

// creates an azure function, with secrets stored in the key vault
module azureFunctions_api 'functionApp.bicep' = {
  name: 'functions-app-deployment-api'
  params: {
    appName: appName
    appInternalServiceName: 'api'
    appNameSuffix: appSuffix
    appInsightsInstrumentationKey: logAnalytics.outputs.instrumentationKey
    keyVaultName: keyvault.outputs.keyVaultName
    msiRbacId: msi.outputs.id
  }
  dependsOn: [
    keyvault
    logAnalytics
  ]
}

As we can see in the code above, bicep scripts are maintainable and we can generally understand the code at a first glance.

Dealing with permissions

Finding your Azure user details before running the deployment

As we would expect, it's necessary to configure some permissions to create or update resources on your Azure subscription. This is necessary for any IaC tool. After we decide who will run the deployment, we must get its ObjectId in Azure AD and pass it as a parameter to the deployment script. So, before trying to run the script, you must first decide who will be the AAD user (not service principal) to run this deployment. For example, to give yourself access to this deployment, run:

    az ad signed-in-user show

You will see many properties. Grab the objectId property. The deployment script will give you adequate permissions to run this deployment.

Working with roles and role assignments - Azure RBAC

The main reason we should consider Azure RBAC is the fine-grained control of permissions we can configure for Azure resources. Also, in the ideal world, with Azure RBAC enabled, we don't need to store keys or secrets across the components of the solution. Azure uses Roles, Role Assignments and Identities to control who can perform specific actions on Azure. Please note that not all the Azure services support RBAC authentication yet at the time of this writing (January 2022). Also, please note that some services have "management platform" permissions and "data" permissions. Commonly, a role can be enabled to only manipulate a service but not see the data stored in it. This is true for Cosmos DB and Azure Key Vault.

In our example, we create an Azure Function that connects to Azure Key Vault without the need to configure the Key Vault-specific role management system as we used to do before. The Key Vault with RBAC enabled delegates authorization to Azure itself. No need for custom code, Connection Strings handling or manipulation of secrets stored somewhere. To allow a service to communicate with others, we must configure who can do what beforehand. This process is not trivial and usually requires trial and error as well as reading the documentation for each role defined for the service we want. Every service will have different permissions available to be used. After we discover the roles needed, we assign them to a Managed Identity. The managed identity is a badge that we manually add to the service (in this sample, the Function App). This badge will contain the role assignments we want - for example, view a secret value. The caller service doesn't have to be changed - just present this badge to the destination service. So, any Azure service that runs under this Managed Identity will have the power to view a secret value. In our example, we want the Azure Functions to have access to the contents of the secrets.

So we have to:

Create a role definition (what we can or can't do);
Create a managed identity (the "badge" itself);
Create a role assignment - give this badge to the service that will call other Azure services.

There are two different kinds of Managed Identities:

User-Assigned Managed Identity: is a Service Principal (an Azure AD identity representing services/custom apps) that can be created before independently of the existence of other services and can be later attached to the services we create. This is ideal in our scenario because we can create the identity, configure it and assign it to the services we want. In our sample, we create an identity and pass it as a parameter to the Function App during its creation. Also, we have the option to do the assignment manually, using Azure CLI. Another important aspect is that the relationship is many-to-many between MSIs and Services. One service can consume multiple user-assigned identities; a given user-assigned identity can be attached to multiple services.
System-Assigned Managed Identity: We also have the option to create an Azure service with a Service Principal that is tied to the lifecycle of the service. For example, we can create a System-Assigned Managed Identity for the Function App and give it the same role assignments during the creation of the service, or later, via deployment script or Azure CLI commands. In general, it's more interesting to use User-Assigned identities because it can be managed as a first-class citizen of our deployment and it appears on the resource group - it's not hidden inside AAD. Also, as you might expect, there is a one-to-one relationship between the system-assigned managed identity and its underlying service.

Creating a suitable role definition that works with Azure Key Vault

The permissions on Azure Key Vault in my opinion are very complicated to set up at the time of this writing. It can be very tricky to discover what is the correct role or why the access to Key Vault is not working. We might also be misguided by many posts saying that the RBAC roles can take some time to propagate and think that it would be potentially a matter of time to have everything working.

To allow the MSI or even the the deployment operator/developer to perform operations on Key Vault, the script defines a new role called "Operator role for the {app name}" with the following definition:

resource roledefinition_deploymentOperator 'Microsoft.Authorization/roleDefinitions@2018-07-01' = {
  name: guid('deployment-operator', appName)
  properties: {
    roleName: 'Operator role for app ${appName}'
    description: 'This role orchestrates this deployment and allows the communication between the components in this solution.'
    assignableScopes: [
      resourceGroup().id
    ]
    permissions: [
      {
        actions: [
          'Microsoft.Authorization/*/read'
          'Microsoft.Insights/alertRules/*'
          'Microsoft.Resources/deployments/*'
          'Microsoft.Resources/subscriptions/resourceGroups/read'
          'Microsoft.Support/*'
          'Microsoft.KeyVault/checkNameAvailability/read'
          'Microsoft.KeyVault/deletedVaults/read'
          'Microsoft.KeyVault/locations/*/read'
          'Microsoft.KeyVault/vaults/*'
          'Microsoft.KeyVault/operations/read'
        ]
        dataActions:[
          'Microsoft.KeyVault/vaults/*'
        ]
        notActions: []
        notDataActions: []
      }
    ]
  }
}

Since I couldn't find any combination of roles that allows the app to read keys and secrets at the same time, I took the path of creating a custom role definition here. This role definition is fairly permissive on Azure Key Vault, but of course, you can make it more restrictive to suit your needs. This Custom Role is defined on the Resource Group scope. This is important because this custom role can be used to control not only Key Vault but all the other services we would want. All the resources created on a resource group inherit from higher scopes' permissions. We can define permissions at the resource group level, subscription or management group level as well as on the resource level itself. As said above, we do the role assignment at the Resource Group scope only and we avoid setting permissions on specific objects. To be clear, we don't assign permissions on the Key Vault itself - we let it inherit permissions defined at the resource group level.

Fig 9: Permissions for the two identities defined at the Resource Group level.

In this sample, the use of Azure Key Vault keys is not really needed. We could potentially go ahead and assign an existing Built-In role like Key Vault Secrets Officer Built-In Role to work only with secrets. Anyway, I decided to create a custom role because it can make things very flexible and all the permissions defined are very clear. In the future, if keys are needed, we won't have more surprises.

Fig 10: Role definition for our Custom Role.

The documentation containing all the available Built-In roles is here: Available RBAC Roles for Key Vault. To reference any of these roles, find the JSON definition of the role right below the description and get the name property of the role, that is a Guid. For example, the role id for Key Vault Secrets Officer is b86a8fe4-44ce-4948-aee5-eccb2c155cd7. You would pass this value directly as a parameter of the deployment script.

Role Assignments on Bicep script

The code to assign the role definition above to a service principal is like this:


// operatorSetup.bicep, called by main.bicep

@description('principalId of the user that will be given the permissions needed to operate this deployment.')
param operatorPrincipalId string

@description('Application name, used to compose the name of the role definitions.')
param appName string
var roleAssignmentName = guid(resourceGroup().id, roledefinition_deploymentOperator.id, operatorPrincipalId, appName)

// assigns the role definition above to the user who will perform the deployment.
resource keyvault_roleAssignment_deploymentOperator 'Microsoft.Authorization/roleAssignments@2020-08-01-preview' = {
  name: roleAssignmentName
  scope: resourceGroup()
  properties: {
    roleDefinitionId: roledefinition_deploymentOperator.id // remember that this the user who is doing the deployment!
    principalId: operatorPrincipalId
  }
}

output roleId string = roledefinition_deploymentOperator.id 
output roleName string = roledefinition_deploymentOperator.name // just for troubleshooting if you need it
output roleType string = roledefinition_deploymentOperator.type // just for troubleshooting if you need it

The result can be seen visually here:

Fig 11: Role assignments at the resource group level.

It creates the assignment at the resource group level as well. We output the role definition id back to main.bicep so it can reuse this id and assign the operator role to the MSI we create. I am using the same role definition for the operator and the managed identity for simplicity. Please note that we can troubleshoot bicep deployments by creating output variables. Each output variable from external bicep files should be should also be "outputted" again in main.bicep as well so the results can be printed on the command line. Please also note that each file can be deployed independently, as long as you pass the correct parameters. When the deployment succeeds, the results are also available in the command line, as well as a very descriptive log on the command line. You can also see the results of the deployment on Azure.

Please note again that I assign, for simplicity, the same role for the deployment operator (in this case, it's me) and the MSI. This is useful for troubleshooting the deployment. For example, Key Vault does not directly allow you to see the keys without specific permission for doing so. Even if Key Vault Administrator or Contributor roles are selected, we won't be able to see the keys, just manipulate the service itself. It is really necessary to give specific permissions to see the contents of a key. This restrictive style of permissions used in Key Vault can cause a lot of surprises during the deployment and further configuration of the solution. This will usually mean that you can have deployment issues because when you forget to configure explicit permission to create or read the keys. The errors are misleading. Even worse: if you don't configure properly the permissions to read secrets, your function app won't even be able to start. In our sample, I also store the storage secrets inside Key Vault. Azure Functions needs a valid storage connection to start. Issues of this kind can be a nightmare and take a lot of time to fix. So please: do take special attention to the permissions you set for users and identities when dealing with RBAC-enabled Key Vaults.

Create a deployment and run on an existing resource group

In this sample, I first create manually a resource group on Azure to host the deployed items. With the resource group created, you go ahead and run this command:

az deployment group create --resource-group sample-app-blog-post-eastus2 --template-file .\src\main.bicep --parameters deploymentOperatorId=aaaabbbb-ccdd-ee11-2233-444455667788

The command needs the ObjectId of the user who will do the deployment and will also need the resource group name. In this case, I am using sample-app-blog-post-eastus2 as an example.

Please note that when you first execute the deployment, the process will take 10 minutes or more because it creates the Cosmos DB Account, which is usually not so fast. Since this deployment uses predictable names for resources, it can be run multiple times. In general, it will update or create the resources defined on each bicep file. In my experience, the subsequent runs take less time since it does not recreate everything. Also, please note that you can pass the --parameters multiple times. Change the variable appName to fit your needs inside main.bicep.

Fig 12: Deployments at the resource group level.

Fig 13: Details of a deployment entry.

Creating a Key Vault instance

The creation of an Azure Key Vault with RBAC enable is much simpler. No need to set custom permissions or so here.

// keyvault.bicep, called by main.bicep
param appName string

@maxLength(24)
param vaultName string = '${'kv-'}${appName}-${substring(uniqueString(resourceGroup().id), 0, 23 - (length(appName) + 3))}' // must be globally unique
param location string = resourceGroup().location
param sku string = 'Standard'
param tenantId string // replace with your tenantId

param enabledForDeployment bool = true
param enabledForTemplateDeployment bool = true
param enabledForDiskEncryption bool = true
param enableRbacAuthorization bool = true
param softDeleteRetentionInDays int = 90

param networkAcls object = {
  ipRules: []
  virtualNetworkRules: []
}

resource keyvault 'Microsoft.KeyVault/vaults@2021-06-01-preview' = {
  name: vaultName
  location: location
  properties: {
    tenantId: tenantId
    sku: {
      family: 'A'
      name: sku
    }

    enabledForDeployment: enabledForDeployment
    enabledForDiskEncryption: enabledForDiskEncryption
    enabledForTemplateDeployment: enabledForTemplateDeployment
    softDeleteRetentionInDays: softDeleteRetentionInDays
    enableRbacAuthorization: enableRbacAuthorization
    networkAcls: networkAcls
    // note: no need for access policies here!
  }
}

output keyVaultName string = keyvault.name
output keyVaultId string = keyvault.id

We need to output the reference of KeyVault to the deployments that will need to use this recently created instance of Key Vault. Please note that this instance is usable immediately and you don't need to wait to store keys and secrets. If you have issues, probably you should tune the permissions for the user/identity that will use this Key Vault.

Tip: Activate Application Insights on Key Vault to facilitate the investigation of access issues. This is crucial.

Creating a Cosmos DB account, storing secrets and configuring containers

It's very cool that we can deploy the account and configure the containers as well in the same execution. We don't need to further manipulate CosmosDB with scripts in this initial deployment.

The whole script with the Cosmos DB deployment is here:

@maxLength(30)
param cosmosAccountId string
param location string
param cosmosDbName string
param keyVaultName string

param tags object = {
  'deploymentGroup': 'cosmosdb'
}

resource cosmosAccount 'Microsoft.DocumentDB/databaseAccounts@2021-10-15' = {
  name: toLower(cosmosAccountId)
  location: location
  kind: 'GlobalDocumentDB'
  properties: {
    publicNetworkAccess: 'Enabled'
    enableFreeTier: false
    databaseAccountOfferType: 'Standard'
    consistencyPolicy: {
      defaultConsistencyLevel: 'Session'
      maxIntervalInSeconds: 5
      maxStalenessPrefix: 100
    }
    locations: [
      {
        locationName: location
        failoverPriority: 0
        isZoneRedundant: false
      }
    ]
    capabilities: [
      {
        name: 'EnableServerless'
      }
    ]
  }
}

resource cosmosDb_database 'Microsoft.DocumentDB/databaseAccounts/sqlDatabases@2021-06-15' = {
  name: '${cosmosAccount.name}/${cosmosDbName}'
  tags: tags
  properties: {
    resource: {
      id: cosmosDbName
    }
  }
}

resource container_leases 'Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers@2021-06-15' = {
  name: '${cosmosDb_database.name}/leases'
  tags: tags
  dependsOn: [
    cosmosAccount
  ]
  properties: {
    resource: {
      id: 'leases'
      partitionKey: {
        paths: [
          '/id'
        ]
      }
    }
  }
}

resource container_employees 'Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers@2021-06-15' = {
  name: '${cosmosDb_database.name}/Employees'
  tags: tags
  dependsOn: [
    cosmosAccount
  ]
  properties: {
    resource: {
      id: 'Employees'
      partitionKey: {
        paths: [
          '/EmployeeId'
        ]
      }
      uniqueKeyPolicy: {
        uniqueKeys: [
          {
            paths: [
              '/EmployeeId'
            ]
          }
        ]
      }
    }
  }
}

module setCosmosConnectionString 'setSecret.bicep' = {
  name: 'setCosmosConnectionString'
  params: {
    keyVaultName: keyVaultName
    secretName: 'CosmosDbConnectionString'
    secretValue: cosmosAccount.listConnectionStrings().connectionStrings[0].connectionString
  }
}

output cosmosAccountId string = cosmosAccountId

The code for setSecret.bicep is:

param keyVaultName string
param secretName string
param secretValue string

resource secret 'Microsoft.KeyVault/vaults/secrets@2021-06-01-preview' = {
  name: '${keyVaultName}/${secretName}'
  properties: {
    value: secretValue
  }
}

In this case, we are creating a Cosmos DB Account using the Serverless mode. I tried to use the cheapest settings here, but it will depend on the workload your project needs. Of course, you can opt-in for the free tier if you have space in your subscription - they allow only one account with the free tier enabled.

It would be difficult to figure out how to build this script from scratch. It is. To discover the settings, I googled many samples and have gone through trial-and-error. There are many samples for this specific for Cosmos, but in general, you should first create want you want manually and try to induce the parameters of the deployment. The Azure Portal allows you to export the deployment template before and after the deployment. You can also go to the upper right corner of the resource you want to script and click JSON view to see how a given resource is defined. You will notice with some experience that many items are the defaults and don't need to be scripted on your bicep deployment script.

For Cosmos we do the following process:

Create the account
Create a new database
Create the leases container (useful for Azure Functions + Change Feed)
Create the Employees container, with an adequate partition key and a unique index.
Save the connection string on Key Vault as a secret, using the Key Vault name passed in as a parameter. Please note that we also use a special function to retrieve the connection string on the fly - don't need to see the value of the connection string.

What about RBAC for Cosmos DB?

RBAC is also available and is the recommended way to work with Cosmos as we can see on this document: Setup RBAC - Cosmos DB.

I have decided to not use this yet because, to be succinct, Azure Functions does not support yet RBAC-enabled Cosmos DB bindings for Consumption Plans. If we use an RBAC-enabled Cosmos DB account, we would need to write code and manage a static instance of a Cosmos DB client. Since, in my humble opinion, the whole point of using Azure Functions + Cosmos is the brilliant integration via bindings and triggers, I decided to avoid this way until this is available for all plans. The details, at the time of this writing, are here: Cosmos DB Extensions 4x and Higher.

Please also note that Durable Functions won't support RBAC enabled connections for now. See information here: Identity-based connections for Azure Functions

Of course, your requirements will be different and using RBAC for Cosmos can be a viable option for your needs. It's available and works well.

Creating the Function App

The creation of the web app designed to work as an Azure Functions app is quite complex. It's simplified on the portal, but on the script, it is somewhat complex. I have also googled this a lot, tried many combinations and finally have a good combination of settings that will just work for most of the common needs.

Log Analytics Workspace + Application Insights

I delegate the construction of these resources to its file since this rarely changes once it succeeds.


// logAnalytics.bicep, called by main.bicep

param appName string
param location string = resourceGroup().location
var logAnalyticsWorkspaceName = 'log-${appName}'
var appInsightsName = 'appi-${appName}'

// Log Analytics workspace is required for new Application Insights deployments
resource logAnalyticsWorkspace 'Microsoft.OperationalInsights/workspaces@2021-06-01' = {
  location: location
  name: logAnalyticsWorkspaceName
  properties: {
    retentionInDays: 30
    features: {
      enableLogAccessUsingOnlyResourcePermissions: true
    }
    sku: {
      name: 'PerGB2018'
    }
    publicNetworkAccessForIngestion: 'Enabled'
    publicNetworkAccessForQuery: 'Enabled'
  }
}

// App Insights resource
resource appInsights 'Microsoft.Insights/components@2020-02-02' = {
  name: appInsightsName
  location: location
  kind: 'web'
  properties: {
    Application_Type: 'web'
    WorkspaceResourceId: logAnalyticsWorkspace.id
    publicNetworkAccessForIngestion: 'Enabled'
    publicNetworkAccessForQuery: 'Enabled'
  }
}

output instrumentationKey string = appInsights.properties.InstrumentationKey

We return the instrumentation key, necessary to connect the Function App to Application insights. I also tried to use the cheapest settings. But keep in mind that the amount of data can significantly increase the costs of the solution! Be careful.

Defining the Function App deployment

The script to deploy the Function app is as follows:


// functionApp.bicep, called by main.bicep

// note: https://docs.microsoft.com/en-us/azure/cloud-adoption-framework/ready/azure-best-practices/resource-abbreviations
param location string = resourceGroup().location
param functionRuntime string = 'dotnet'

@description('A name for this whole project, used to help name individual resources')
param appName string

@description('The name of the role or service of this function. Example: Api CommandHandler, EventHandler')
param appInternalServiceName string

@description('Id of a existing keyvault that will be used to store and retrieve keys in this deployment')
param keyVaultName string

@description('User-assigned managed identity that will be attached to this function and will have power to connect to different resources.')
param msiRbacId string

@description('Application insights instrumentation key.')
param appInsightsInstrumentationKey string
param deploymentDate string = utcNow()
param appNameSuffix string

var functionAppName = 'func-${appName}-${appInternalServiceName}-${appNameSuffix}'
var appServiceName = 'ASP-${appName}${appInternalServiceName}-${appNameSuffix}'

// remove dashes for storage account name
var storageAccountName =  toLower(format('st{0}', replace('${appInternalServiceName}-${appNameSuffix}', '-', '')))

var appTags = {
  AppID: '${appName}-${appInternalServiceName}'
  AppName: '${appName}-${appInternalServiceName}'
}

// Storage Account - I am using 1 storage account for each function. It would be potentially shared across many function apps.
resource storageAccount 'Microsoft.Storage/storageAccounts@2019-06-01' = {
  name: storageAccountName
  location: location
  sku: {
    name: 'Standard_LRS'
  }
  kind: 'StorageV2'
  properties: {
    supportsHttpsTrafficOnly: true
    minimumTlsVersion: 'TLS1_2'
    encryption: {
      services: {
        file: {
          keyType: 'Account'
          enabled: true
        }
        blob: {
          keyType: 'Account'
          enabled: true
        }
      }
      keySource: 'Microsoft.Storage'
    }
    accessTier: 'Hot'
  }
  tags: appTags
}

// Blob Services for Storage Account
resource blobServices 'Microsoft.Storage/storageAccounts/blobServices@2019-06-01' = {
  parent: storageAccount

  name: 'default'
  properties: {
    cors: {
      corsRules: []
    }
    deleteRetentionPolicy: {
      enabled: true
      days: 7
    }
  }
}

resource keyVault 'Microsoft.KeyVault/vaults@2019-09-01' existing = {     
  name: keyVaultName
  scope: resourceGroup()     
} 

module setStorageAccountSecret 'setSecret.bicep' = {
  name: 'stgSecret-${appInternalServiceName}-${deploymentDate}'
  params: {
    keyVaultName: keyVault.name
    secretName: '${storageAccount.name}-${appInternalServiceName}-ConnectionString'
    secretValue: 'DefaultEndpointsProtocol=https;AccountName=${storageAccount.name};EndpointSuffix=${environment().suffixes.storage};AccountKey=${storageAccount.listKeys().keys[0].value}'
  }
}

// App Service
resource appService 'Microsoft.Web/serverfarms@2020-12-01' = {
  name: appServiceName
  location: location
  kind: 'functionapp'
  sku: {
    name: 'Y1'
    tier: 'Dynamic'
    size: 'Y1'
    family: 'Y'
    capacity: 0
  }
  properties: {
    maximumElasticWorkerCount: 1
    targetWorkerCount: 0
    targetWorkerSizeId: 0
  }
  tags: appTags
}

// Function App
resource functionApp 'Microsoft.Web/sites@2020-12-01' = {
  name: functionAppName
  location: location
  identity: {
    type: 'SystemAssigned, UserAssigned'
    userAssignedIdentities: {
      '${msiRbacId}': {}
    }
  }
  kind: 'functionapp'
  properties: {
    keyVaultReferenceIdentity: msiRbacId
    enabled: true
    hostNameSslStates: [
      {
        name: '${functionAppName}.azurewebsites.net'
        sslState: 'Disabled'
        hostType: 'Standard'
      }
      {
        name: '${functionAppName}.scm.azurewebsites.net'
        sslState: 'Disabled'
        hostType: 'Standard'
      }
    ]
    serverFarmId: appService.id
    siteConfig: {
      appSettings: [
        {
          name: 'AzureWebJobsStorage'
          value: '@Microsoft.KeyVault(VaultName=${keyVault.name};SecretName=${storageAccount.name}-${appInternalServiceName}-ConnectionString)'
        }
        {
          name: 'WEBSITE_CONTENTAZUREFILECONNECTIONSTRING'
          value: '@Microsoft.KeyVault(VaultName=${keyVault.name};SecretName=${storageAccount.name}-${appInternalServiceName}-ConnectionString)'
        }
        {
          name: 'APPINSIGHTS_INSTRUMENTATIONKEY'
          value: appInsightsInstrumentationKey
        }
        {
          name: 'APPLICATIONINSIGHTS_CONNECTION_STRING'
          value: 'InstrumentationKey=${appInsightsInstrumentationKey}'
        }
        {
          name: 'FUNCTIONS_WORKER_RUNTIME'
          value: functionRuntime
        }
        {
          name: 'FUNCTIONS_EXTENSION_VERSION'
          value: '~3'
        }
        {
          name: 'CosmosDbConnectionString' 
          value: '@Microsoft.KeyVault(VaultName=${keyVault.name};SecretName=CosmosDbConnectionString)'
        }
      ]
      use32BitWorkerProcess: true
    }
    scmSiteAlsoStopped: false
    clientAffinityEnabled: false
    clientCertEnabled: false
    hostNamesDisabled: false
    dailyMemoryTimeQuota: 0
    httpsOnly: false
    redundancyMode: 'None'

  }
  tags: appTags
}

What we do:

Create the storage account, needed by Azure Functions runtime. I am creating one storage for each function. Of course, you can use existing storage, share it with many functions.
Enable blob services for the account - seems to be required.
Get a reference of an existing key vault instance and expose it to the script. Note that we require an existing instance of Key Vault. I decided to do this to show an example of how we can reference existing resources on scripts.
Save the storage account connection string on the key vault
Create an App Service designed to be used on a Consumption plan (Y1 Dynamic). The App Service is like a server where the functions will run.
We define the Function App, using the Managed Service Identity (MSI) that we defined before. We use some defaults and we configure the settings using a special syntax for this.

There are many settings that you can investigate and try yourself. In the next picture, on the JSON View tab, you can see how complex the function can be:

Fig 14: Function App definition seen when JSON View is clicked. Many settings don't need to be set, but some of them can help to troubleshoot failed vs successful deployments. Almost all resources will have a JSON definition - go to Overview tab and find it in the upper-right corner.

The script is finished. Just run and it will work.

Function App Identity configuration

The script above defines the identity for the function like this:

identity: {
    type: 'SystemAssigned, UserAssigned'
    userAssignedIdentities: {
      '${msiRbacId}': {}
    }
  }

There are two properties: type and userAssignedIdentities.

I have specifically tried all the combinations possible for the type parameter. It is required to enable the SystemAssigned identity if we only want to use the UserAssigned identity. I lost hours of hard troubleshooting simply trying to understand why the app doesn't start when only the UserAssigned identity is used. I tried to find deployment/permission issues using the Diagnose and solve problems wizard and never found a solution. The only way I was able to discover the issue was to enable Application Insights for the Key Vault and see the errors regarding the access. Then I had to get the objectId printed in the log and had to look for any identity with that id. It turns out that for some reason, the User Assigned Id was not able to reach the Key Vault and no error was thrown. Allowing the System Assigned Id enabled the function app to reach the Key Vault and finally, throw the exact exception describing the missing permissions when the Function App tried to access the secrets. I know this can be very confusing to explain in words, but it seems that we need to enable both types of identities to touch the Key Vault. I will try to simulate some problems and troubleshooting steps in a further post.

The property userAssignedIdentities is badly documented and it was very hard to find the correct syntax to configure this. I also spent many hours on this. The value of the MSI object id is passed as a parameter here. This MSI will carry the permissions we configured above and will try to interact with Key Vault. I hope you don't find issues with this. If so, again, please enable monitoring for your Key Vault instance and find issues on the logs.

Syntax to reference Key Vault settings inside Function App Configs

It's very cool to have the ability to call the Key Vault directly from the configuration of the Function app. This is especially nice because there are no code changes to load secrets and expose them to the app - everything is done on your behalf.

The syntax is:

'@Microsoft.KeyVault(VaultName=kv-some-keyVault;SecretName=CosmosDbConnectionString)'

More details are described here: Key Vault references in configuration.

Application Insights' Instrumentation Key

We usually don't need to consider the Instrumentation Key as a secret. Many DevOps teams save this key on Key Vault just for convenience. In our case, I copied the value into the configuration settings directly.

Useful commands

Purge an Azure Key Vault instance after your delete it. Probably you will need if you retry this deployment many times:

az keyvault purge -n kv-sampleApp-1234 --no-wait

Get the current logged-in information:

az ad signed-in-user show

Grab a list of current role definitions:

az role definition list > roleDefs.out

Grab a list of current role assignments and output to a file:

az role assignment list > roleAssignments.out

Delete role assignment by hand

az role assignment delete --id "/subscriptions/xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/providers/Microsoft.Authorization/roleAssignments/<object id of the identity>"

Final hints

Go to the Deployments menu on the resource group and open each of the deployments that failed to see the real reason for a given failure. The errors on the VSCode are sometimes very misleading or hard to read because it tries to concatenate many error messages.
Application insights is your friend when dealing with Key Vault issues. To enable it, you have to go to Monitoring > Diagnostic Settings > + Add diagnostic setting. Ask to flow Audit Events and AzurePolicyEvaluationDetails to the Log Analytics workspace defined in this deployment.
When a function does not start, it's very hard to troubleshoot because the application is not available. No Kudo. No console. No logs in Application Insights. No Diagnose and Solve Problems option will work. It's crucial to be sure that function identity is working.
The development team might need the same or similar permissions to run the solution on their local machines.

Conclusion

I tried to show how we would structure a realistic deployment script using Bicep. I believe that for Azure-only scenarios this is the way to go. The script code is composable, easy to read and maintainable. There are also native functions that help in many aspects.

To structure a full application to be deployed, an important amount of time, trial-and-error, patience and troubleshooting skills is necessary. I recommend that you consider splitting all the deployments into smaller scripts and try to deploy each one individually to isolate deployment issues.

This is the beginning of the journey. With a working script in hands, it's important to make this work on the deployment pipeline and add code to this solution. I will cover this in the future as well.

Thanks for reading!
Mário

Written with StackEdit.

Serverless Event-Driven Architecture on Azure: A Worked Example - Part 1

Mario Meyrelles — Fri, 12 Jun 2020 21:14:43 +0000

Introduction

The idea behind this series post is to try to demonstrate a somewhat realistic example of how to structure a back-end architecture using simple Azure serverless offers. This project will implement a realistic business application with extensive use of CQRS and Event Sourcing concepts as well as some more advanced patterns like a realistic implementation of a Process Manager. Most components of this architecture are designed with reactivity in mind, in the sense that each component is called during a reaction to a formal message arriving on the network. As it will be explained later, Azure provides excellent tools to help you leverage reactive programming on the cloud. This architecture is inherently asynchronous and message-based.

We will try to show a viable way to start new projects on the cloud using serverless products from Azure. Capitalizing on the serverless offers of Azure can be a huge time saver and cost saver to teams that need to rapidly create modern, scalable, and elastic solutions. This is particularly interesting for startups and teams wanting to deliver flexible and easy to understand solutions on the cloud;

This article will give you directions to help you achieve a viable serverless event-driven, reactive architecture on Azure. We will strive to inspire your team to consider a serverless approach to your next project.

What will be covered in this series

We try to explain the following concepts and techniques:

Motivation of why event-driven architecture is an advantage in the cloud
Example of CQRS, Commands, Events, Command Handlers implementations
Reactive techniques on Azure
Using CosmosDB as a genuine Event Store of the whole system. We demonstrate the concept of an event store, global state, projections and what to do when an event arrives
Propagation of events to other consumers and yet unknown consumer using Event Grid and a worked example of a service that reacts to external events and depends on manual intervention
Detailed implementation of a Process Manager that orchestrates a process and maintains state on the cloud
Examples of retry logic and Dead Lettering of messages
Deep usage of Azure Functions and Durable functions
Monitoring using Application Insights and troubleshooting of problems

What will not be covered in this series

To avoid the excessive complexity of this example, won’t be covered:

API Management, Logic Apps, Service Bus and other related offers as we wanted to keep things simple
Event storming and DDD modeling techniques in detail
Authentication
Unit Testing techniques
Pipelines of Continuous Integration or Continuous Deployment, Terraform or ARM
Front-end code. We have included a simulator that can prepare the system for the desired scenarios using a console application.
Reporting
Data and AI integrations
GraphQL and gRPC

We plan to cover some of these items in the future.

An asynchronous, event-driven strategy for cloud-first development

Synchronous systems are systems where the reaction to actions (started or not by the user) are immediately processed, persisted, and returned to the requestor. This is a very natural and common way to implement common business applications and we have seen this kind of implementation everywhere. Monoliths are normally implemented using synchronous strategies, usually, with a central database to store the global state of the whole application. Reads and writes happen in the same location, competing for network and storage resources. Improvements in the storage resources are hard to achieve and the database must be very optimized to deliver good results. Database scaling is frequently expensive.

Since the global state is shared it is often difficult to implement or modify existing tables or fields in the database. It`s always complicated to find the systems impacted by a single change. Adding bigger features that require deeper changes in table structures is often impossible or very hard to implement without long negotiations. Another disadvantage of monolithic architectures is the possibility of resource starvation, timeouts, and failure of integrations to 3rd party systems with the potential affect the critical path of the application. User experience can easily be degraded. The whole system becomes more sensitive to issues during the processing of the requests. These scenarios are the major motivation to migrate to microservices, along with the possibility to divide the development work among many independent teams. There is a common expectation that just the migration to a microservices architecture is enough to achieve a more flexible architecture and more agile software development and delivery process. Not necessarily true.

To fully use the potential of the cloud services, we should separate the system in components that can do well one task. A monolithic approach would not work when we must scale different parts of the system at different times. Separating the system in smaller pieces will lead us to design the system that inevitably requires us to use microservices approach to achieve success. We design thinking in services that can execute individual work with minimum dependencies on other parts of the system. Each service can be scaled individually, can fail, can be stopped, and can be changed with very limited effects on the other parts of the system. Potentially, the maintenance of one part should not affect any other part.

We also must be aware that using a serverless approach, we can have the benefit of only being charged proportionally to use and we can decommission parts of the system that don`t need to be active at all times. Another advantage is that many services are billed by the number of messages processed. On Azure, many services are free for the first thousands or millions of messages processed per month. This is an important cost saver compared to more traditional approaches or even, container orchestrator approaches that imply a fixed minimum cost to run the business due to the need to have virtual machines running 24 x 7.

An event-driven approach is well suited for microservices communication in the cloud. We embrace asynchronism. With the use of messaging it is possible to separate producers and consumers so we can scale them individually. In the context of event-driven architectures, we usually classify messages as Commands, Queries, Events, and Notifications. We tend to see the dynamics of the system as a sequence of commands and events ordered in time. Consequently, modern systems can often be understood as a stream of events. Azure has many offers that can help in business apps, IoT, Data+AI, and more. We cover in this example a realistic use case that uses an event-driven strategy to deliver a business application.

Doing event-driven architecture on the cloud has a very interesting consequence: the cloud itself can work as a reactor. The cloud assumes the responsibility to take a given message and deliver it to its handler without the consumer knowing anything about how the input arrived. The consumers in this kind of architecture passively receive the messages and react to them. A reactive service does not know anything about the messaging infrastructure, does not have code related to messaging itself. Reactive services are simpler to understand and develop.

DDD is a very important tool to design cloud-native systems. In the world of distributed systems, it's very difficult to manage business rules, to manage transactional boundaries, and to understand the big picture. We must design the systems considering commands, events, and bounded contexts from Day 0. The development of the application should anticipate the domain model from the beginning and be prepared for the evolution of the project.

Architectural Overview

We will cover the implementation using Azure serverless offerings and we will go in deep detail to illustrate how anyone can start using an event-driven architecture.

We are extensively using Cosmos DB, Azure Functions, and Event Grid as the main cloud components of our architecture. We also use in this example storage accounts and Durable Functions. Azure Functions is the best way to run your code on the cloud when it`s needed to react HTTP calls, changes in Cosmos DB collections, scheduled calls, and more. Azure Functions offers a sophisticated structure of bindings and triggers that allows Azure Functions to be an intelligent glue for various Azure services. A function must execute fast and should not maintain state. Using a stateless approach, we can easily create functions that can be scaled proportionally to the number of calls that need to be processed. That means that Azure can create multiple instances of a function and consume many messages as needed. This is a very important reason to keep parts of the process separated in different functions and potentially, different function apps.

Cosmos DB is often viewed as an expensive way to store data on the cloud. But there have been lately many efforts to reduce the costs. Use cases with a smaller number of RUs can now be favored. Cosmos DB, as we are speaking in April/May 2020, can now use database-level pricing instead of collection-level pricing. Now it’s possible to share RU/s and reserve RU/s for specific collections when necessary. This is a very important price reduction as not all collections have intensive read/write needs. Since March 2020, Cosmos DB offers a free tier for the first 400 RU/s and 5GB in the month and lets you share the free tier with 25 containers of the database. This is a huge cost saver for small use cases. Smaller microservices can now consider the use of Cosmos DB as their primary storage engine. We do this in this project.

When starting a project from scratch, there is always room to analyze if the storage model should be relational or non-relational. Of course, it depends a lot on the workload of the app to be created. But in general, in the industry, we usually see that startups tend to adopt the NoSQL model because it’s far easier to design a document database than a complete relational model. Startups are commonly discovering a market fit and are changing and adding features to their products.

In this example, we don`t make use of a relational database, even though there is an Azure offering for SQL Server that is serverless. We stick with Cosmos DB because we are creating a project from scratch that does not need a relational database and all its complexity. Another important feature that Cosmos DB offers is the Change Feed. This feature is a key enabler of event-driven architectures. The Change Feed calls an Azure Function when a new item is added or modified on a given container. This is nice because a function now can be reactively called by an insertion on a Collection. Cosmos DB is a reactor on our system, sending messages in the order they arrive. One or more listeners react to each message arrived. This mechanism is very analogous to Apache Kafka – a producer adds a message to the stream and potentially some consumer processes the message. The message is not deleted after the consumption, enabling a scenario where multiple consumers can process their messages in the speed they want/can process. The Change Feed is just a list of messages that can be read from the beginning at any time – the consumer decides what is the next item to process. Azure Functions automates the reading of the Change Feed and the management of the “cursor” that marks the position of the message in the feed.

An event store is just a database that stores events in an architecture that uses Event Sourcing. Cosmos DB can be seen as a helpful event store in the system, since it stores the messages and can reactively notify Azure Functions about new items added. The Azure Function that reacts to a message on CosmosDB is a projector and optionally, it can update the read model as a reaction to an event. The read model is just a projection of the global state stored on the event store. This means that we can potentially delete the read model and do a reconstruction using the data in the event store. We can even have multiple projections – multiple read models. In our project, we update the read model a message arrives every time.

Event Grid is a serverless offering designed to send individual messages to applications and assure it is really sent or dead lettered. The use case for Event Grid is to be a glue that connects many event sources and event consumers, carrying messages that have much more value and can cause problems if lost. It’s an enabler of reactive scenarios in the cloud. One great aspect is that it connects to Azure Functions and Webhooks. This integration option can allow any app that exposes a webhook to simply be called by Event Grid. For example, if an event handler exposes a webhook, it can readily react to an event published by our system with Event Grid’s help without any modification in code. Of course, we should note that Event Grid can listen to events published internally by many Azure components.

Now let’s explore how the backend of our app is designed. We have an example of a very common scenario where the front-end wants to trigger a change in the global state of the system:

The diagram of the figure above shows what happens when a request is sent from the client. We consider as a client in this scenario any client-side technology like mobile, web, or desktop application. The bullets in the diagram show the steps taken when a user action begins.

Step 1 – Preparation of the request: The client needs to perform an action that is expected to change the state of the system. In our example, the user wants to send some credits to his son. A mobile app will have run a query to obtain the account details of his son. This is done by sending a query to the read side of the application. The read side of the application, in the sense of CQRS, is a database and code optimized for reading that is physically apart from the database used to store data on the "write side". The client app sends the data over the wire to an HTTP endpoint. It`s important to have a read model updated as soon as possible.

Step 2 – Request Validation and Command Construction: The Azure function receives the request and validates the body of the request, validating the structure of the request. More complex validation can require the use of the read model to help. After the validation of the request, the command is constructed and is sent to a Cosmos DB collection. The command is not immediately handled as it would be expected in other architectures. We avoid handling the command inside the function that processes network requests because this regard the HTTP endpoint works as a producer of messages (commands) that will be handled by different command handlers. We decouple the command construction from the command handling in code and deployment dimensions. Another important issue about command processing is the fact that it is possible to receive potentially duplicate commands. The command store does not allow commands with the same hashes and therefore, we are protected from duplications and related issues.

Step 3 – Handling Commands: The commands are saved to a collection. The command handler logic is triggered during a reaction to a command being published on the system. If the command fails, the client should be notified. If the command succeeds, the client should not be notified yet. The event should be published first. In general, each command handler produces one or more events. The command handling should have a retry mechanism since transient failures could result in a command failure that can be avoided with a simple retry.

Step 4 – Persisting events on event store: When the command is processed with success, the command handler publishes on or more events that represent a change in the global state. To be clear, a successful insertion in Cosmos DB Events collection means that this event has become public and can’t be removed or changed anymore. It’s a fact. The event, published on CosmosDB by the mechanism of Change Feed + Azure Functions, can now be processed further.

Step 5 – Primary Event Handlers: The only way to process documents using the Change Feed using code and serverless technology is by using an Azure Function. So, it’s mandatory to have an Azure Function listening to events that are published on our event store if we want to react to them. And it’s also the only way to publish Cosmos DB changes to Event Grid. Since this function is mandatory, we call this function a Primary Event Handler. In our project, we also use this function to continuously update the read model of the application so the client can see the results of their request as soon as possible. We call this a primary reaction. After this primary reaction to this event is completed, we can send the event to Event Grid and let other event handlers work. A primary reaction should always be a crucial use case of the system, usually, when the user must see the results of their actions immediately. In our example, all reactions that required fast change on the user’s balance are done in primary event handlers so the clients can be notified of a change in the balance as soon as possible and therefore, reduce the chance of further command rejections due to incorrect balance. In the next post, we show the details of the business problem and an implementation strategy.

Step 6 – Read Model updated: The read model is a projection of the global state. As said before, the global state is just an immutable sequence of events. Nothing more. It means that the read model of the application can be only be updated after the global state is updated on the event store. The read model can be any kind of database/storage. Also, we can have multiple read models* for different needs. For example, a read model can be a simple collection on Cosmos DB, a SQL Server database, or a sophisticated graph database. Please note that services can maintain and change their local state independently. We will further discuss how services can change the global state in the next post.

Step 7 – Event Grid dispatches events: The idea is to forward all events to Event Grid so, in the future, a new service can be added to listen to events of interest. Event Grid can help promote reactive programming across the cloud, helping in the transmission of events or other kinds of messages to any service. This is very important because Event Grid is an integration point that can allow many systems to connect and exchange messages. This Event Grid can start workflows by calling webhooks, by adding messages to queues and some other ways. Please note that Event Grid is designed to help "real-time" use cases, where the reaction should occur immediately after the message arrives. Event Grid can only route events on the present. New services won't be able to read past events unless they consume Cosmos DB data. It`s not currently possible to schedule messages to the future, although you can do this easily on Azure Service Bus.

Step 8 – Secondary Event Handlers: A secondary event handler in this architecture is an event handler that is called by Event Grid to process an event. It can be another system with an admin GUI, can be a microservice, can be Power BI, can be anything. In our example, we created a web application that manually validates if a given transfer operation is acceptable or not. It`s worth noting that those event handlers can produce more commands and events during a reaction to an event. This is very interesting and we discuss this in detail in the future parts of this series.

Step 9 – Other Reactions: Sometimes the services called during an event reaction will publish new events that potentially will change the global state. This is naturally acceptable. We can still react to these events as part of a more complex choreography of events.

Some Thoughts

Domain logic and serverless architectures

To make things easier, in this example we share a library with all the functions and services. Our domain logic is centralized in a single library, as well as all commands and events of interest. This is a design decision that has pros and cons. Centralization of business rules, as we will see in the next part of this series, can dramatically help to assure we maintain the domain in a consistent state and we don't brake domain invariants. On the other hand, it brings coupling to a programming language and technology. For example, we indirectly force all the functions to be written in .NET Core 3.

The solution should always be serverless?

No. Microservices can be hosted, for example, in a Kubernetes cluster, reacting normally to events and publishing more events. This kind of integration is planned to be in a future post.

Event Processing

As we saw above, we use Cosmos DB as our persistent event storage. This is simple to understand and its reactive capabilities make things simple to process events further. Having a queryable event store is an important advantage and can help in maintaining the overall simplicity of the solution. There are pros and cons to using Cosmos DB. To achieve higher throughputs it's often very expensive and this can be a blocker in many projects. On the other hand, using a managed service like this allow for a planetary-scale event store, with high availability and automatic management of data replication. And usually, it's very expensive to do this kind of architecture in other technologic stacks and clouds. For common business apps and use cases, Cosmos DB can help with reasonable costs. The growth in scale can be controlled, as well as the cost. We can do event streaming even for business applications, which is great.

There are use cases that are far more complex and depend on very fast data ingestion and processing. Azure offers many technologies to help in these scenarios. Notable examples of these offers are Event Hubs, Kafka (AKS / HDInsight), Stream Analytics, Azure Databricks. Azure Functions and Cosmos DB can help in this architecture also. These components can work in combination to implement Lambda Architecture or Kappa Architecture.

As a final note, it`s always important to understand if we are streaming messages/logs/telemetry information to be processed in near real-time or we are streaming business commands and events through the system. The notion of value of each message is very important. For example, it's viable to lose a message informing the temperature of a sensor that is sent every 30 seconds. On the other hand, we would have problems if we lose a message containing business rich information like the submission of a form used to create a new account. Different use cases require different architectures and different investments in cost.

More info on Event Streaming Architecture:
Streaming choices on Azure
Examples of Streaming a Scale on Azure
Kappa Architecture

Kafka/Event Hubs vs Cosmos DB as an Event Store

Kafka was not intended, originally, to store messages forever. Kafka, like Azure Event Hubs, works better for use cases that need to deal with high data ingestion throughput and distribution to multiple consumer groups that can consume these messages at their own pace. The received messages are intended to stay on the log for a configurable time. Handling millions of messages per second, these messages are usually forwarded to real-time analytics providers and usually, batching and storage adapters. Kafka and Event Hubs are not meant to be the final store for the data - they are a very fast and scalable service for data ingestion and normally, the first part of a data processing/analytics pipeline.
That said, data ingested should normally be processed and stored where it can be queried and used in applications. In our opinion, Kafka should not be used as an event store, neither Azure Event Hubs. Products like Cosmos DB, EventStore (https://eventstore.com/) among others are more convenient for this purpose. Cosmos DB is a very interesting option when we are using the Azure cloud because it's serverless and there is no need to worry about its internal management. As a side note, other relational databases can be used also as an event store and many people have done so (https://github.com/commanded/eventstore).

Cosmos DB can help in the scenario of doing CQRS and Event Sourcing because it can be both an ingestion engine and an event store database with great querying support, not to mention the ability to automatically invoke a function when new data arrives. In the next part of this series, we delve into details of the code can be structured to help succeed using Event Sourcing techniques.

Conclusion

In this first part of the series, we introduced how a team can structure the project to create an event-driven business application, accomodating the needs for command and query separation, and taking advantage of an event store for the application. In the next part of the series, we will go into greater detail, explaining the code necessary to handle commands. We will also talk about the aggregate design.

Serverless Event-Driven Architecture on Azure: A Worked Example - Part 2

Mario Meyrelles — Wed, 03 Jun 2020 13:40:15 +0000

Introduction

In the first post of this series here, we discussed how we can design a serverless and event-driven architecture project with components that are fairly easy to understand and integrate. In this second part of the series, we focus on command handling and Aggregate Root design, illustrating how to process commands and generate events inside the business logic. We created a fictional business scenario to help illustrate the concepts.

Some Business Context

Our New Prepaid Card

We are considering in this example a hypothetical prepaid card that stores credits that users can load using their own money. These credits can be used to shop on physical stores using a mobile app.

For users, there is a fee of 0,02% when money is loaded into the card. If the user loads more than $500.00 in a single transaction there is no fee. Users can freely send credits to other users of the platform using the mobile application offered to the users. There are no limits or charges for credits transfers. Transactions involving a significant amount of money are subject to a validation process and can take longer to be processed. The users also can apply for a plastic card if they wish to do credit and debit transactions using the network of the common credit card operators. This card can also be used to do ATM withdrawals. The only costs that will be applied for plastic card transactions are the processing and operator costs, that are not under our control.

For merchants, there is only a 0,025% fee on each transaction for merchants using the platform and absolutely nothing more in the first year. In the next year, the fees will be still low but can be raised a little bit. Any merchant can use this platform to accept payments configuring an account on the platform. The setup process is online and usually, it`s possible to start selling in less than two working days. The merchant can see the money corresponding to each sale in less than 24 hours in their account. The merchant also can receive the payment in credits and make the conversion to money in the future. If the merchant waits 31 days to make the conversion, the transaction fee is removed for the given sale.

The scope of this sample is to model the process of transferring loaded credits from one user to another using the app. We will ignore the other ways to use credits and all the complexities involved. Also, we are ignoring the reload process and we are ignoring the fees management.

Exploring the Domain

As shown above, we have partitioned the problem in a more understandable structure. We have 3 bounded contexts here:

Bounded Context	Mission
User Management	Focus on the lifecycle of a user as a customer of our prepaid card service. In this sample, we will assume that all cardholders are already created, are valid, and have the app configured. We will only seed configured users and we won’t implement any use case.
Balance Management	Controls the cardholder's balance. Any action like a reload, payment or credit transfer is ultimately done via a credit or debit transaction.
App Credits Transfers	Controls free credit transfers done inside the platform. Collaborates with Balance Management to securely add or remove credits from the accounts involved.

As you can imagine, the whole business problem space can be huge. For sake of simplicity we are only showing subdomains of interest in green:

Subdomain	Mission
Account Transactions	Assures that all the requests for debits and credits are fulfilled or rejected correctly, avoiding overdrafts and duplicated transactions. In this implementation, we show the details of working with event sourced aggregates.
Credits Transfers	Responds for all the free credits transfers. The transfers can be done immediately and can be scheduled. Values considered too high are subject to human validation. Automatic rejection of transfers can occur when certain conditions are met. This subdomain shows an implementation of a Process Manager that effectively settles the credits transfers that are accepted after the validation process. We also implement the workflow for human validation in this project.

Subdomain

Mission

Account Transactions

Assures that all the requests for debits and credits are fulfilled or rejected correctly, avoiding overdrafts and duplicated transactions. In this implementation, we show the details of working with event sourced aggregates.

Credits Transfers

Responds for all the free credits transfers. The transfers can be done immediately and can be scheduled. Values considered too high are subject to human validation. Automatic rejection of transfers can occur when certain conditions are met. This subdomain shows an implementation of a Process Manager that effectively settles the credits transfers that are accepted after the validation process. We also implement the workflow for human validation in this project.

Now it's time to go into the details of the aggregates, commands, and events we plan to implement in this sample project:

Aggregate	Commands	Events
Cardholder
Prepaid Card Account (Account Transactions)	DecreaseCredits IncreaseCredits ReverseCreditsChange	CreditsDecreased CreditsDecreaseRejected CreditsDecreaseReversed CreditsIncreased CreditsIncreaseRejected CreditsIncreaseReversed
App Credits Transfer	CreateCreditsTransfer	CreditsTransferAccepted CreditsTransferFailed CreditsTransferManuallyApproved CreditsTransferManuallyRejected CreditsTransferRejected CreditsTransferScheduled (sprint 2) CreditsTransferSentToManualValidation CreditsTransferSucceeded

As the reader can see, the Credits Transfer aggregate is controlling its state using events to represent most of the possible states of a credits transfer. The events in the Prepaid Card Account aggregate are supporting the credits transfer and may not be exhaustive. The Cardholder aggregate only supports the Prepaid Card Account during validation and won`t have any command or event considered in this partition of the platform design. We won’t show in this article how we work scheduled commands and events. This discussion will be done in a future post.

Technical Implementation

Remembering the general architecture of the last post

Before diving into more technical details and code implementation, let's remember the general architecture that we described in the last post:

HTTP Endpoints and Azure Functions

HTTP endpoints are the most common way to communicate with the external world. Often, HTTP endpoints and used inside microservices communication. Azure Functions are very useful to build very lightweight APIs that can be used as a backend. In more complex architectures, the API layer can be put behind Azure API Gateway.

In the context of event-driven architectures, logic inside HTTP endpoints is part of command processing and therefore, can and should help the client to issue correct commands. This means that they can have access to read models and they write to the command store.

Basic API with Azure Functions

The implementation of a typical HTTP that can be like the following:

// TransfersApi.cs
using FluentValidation;
using FluentValidation.Results;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.Documents;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using ServerlessStreaming.DomainModel.CreditsTransferAggregate;
using ServerlessStreaming.DomainModel.CreditsTransferAggregate.Commands;
using System;
using System.IO;
using System.Linq;
using System.Threading.Tasks;

namespace ServerlessStreaming.Api
{
    public static partial class TransfersApi
    {
        [FunctionName("TransfersApi")]
        public static async Task<IActionResult> Run(

            // be sure to add content-type: application/json on your request!
            [HttpTrigger(AuthorizationLevel.Function, "post", Route = "create")] 
            HttpRequest req,

            [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Commands",
                ConnectionStringSetting = "CosmosDBConnectionString")]
            IAsyncCollector<CreateCreditsTransferCommand> commands,

            ILogger log)
        {
            log.LogInformation("Starting validation of Credit transfer request...");

            using var reader = new StreamReader(req.Body);
            string requestBody = await reader.ReadToEndAsync();
            var payload = JsonConvert.DeserializeObject<CreditsTransferModel>(requestBody);

            if (payload == null)
            {
                return new BadRequestObjectResult("Invalid Credit Transfer Request.");
            }

            var validator = new CreditsTransferValidator();
            var validationResult = validator.Validate(payload);

            if (!validationResult.IsValid)
            {
                return new BadRequestObjectResult(validationResult.Errors.Select(e => new {
                    Field = e.PropertyName,
                    Error = e.ErrorMessage
                }));
            }

            var command = new CreateCreditsTransferCommand(
                    aggregateId: payload.Id,
                    correlationId: Guid.NewGuid().ToString(),
                    createdAtUtc: DateTime.UtcNow,
                    payload: payload);

            try
            {
                await commands.AddAsync(command);
                await commands.FlushAsync();

                log.LogWarning($"Request Accepted. CorrelationId: {command.CorrelationId}, AggregateId: {command.AggregateId}");
                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                return new AcceptedResult();

            }
            catch (DocumentClientException ex) when (ex.Error.Code == "Conflict")
            {

                log.LogWarning("Returning Not Modified: ", payload.Id);
                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                return new StatusCodeResult(304);
            }
        }
    }

    public class CreditsTransferValidator : AbstractValidator<CreditsTransferModel>
    {
        // see also: https://www.tomfaltesek.com/azure-functions-input-validation/

        public CreditsTransferValidator()
        {
            RuleFor(x => x.Id).NotEmpty();

            RuleFor(x => x.Sender.Id).NotEmpty();
            RuleFor(x => x.Receiver.Id).NotEmpty();

            RuleFor(x => x.Sender.AccountId).NotEmpty();
            RuleFor(x => x.Receiver.AccountId).NotEmpty();

            RuleFor(x => x.Sender.DocumentNumber).NotEmpty();
            RuleFor(x => x.Receiver.DocumentNumber).NotEmpty();

            // todo: validate if sender and receiver are different
            // todo: validade if account and users are valid

            RuleFor(x => x.Amount).GreaterThan(0m);
        }

        protected override bool PreValidate(ValidationContext<CreditsTransferModel> context, ValidationResult result)
        {

            if (context.InstanceToValidate == null)
            {
                result.Errors.Add(new ValidationFailure("", "Please ensure a model was supplied."));
                return false;
            }
            return true;
        }
    }
}

The code above demonstrates how an Azure Function can react to HTTP requests and produce commands. The function is annotated with bindings and triggers. We can see a trigger carrying the payload of an HTTP request and we can see an output binding that sends the commands to CosmosDB, our command and event store in this project. We can also see an ILogger parameter in the function signature. The logger is extensively used to send data to Application Insights and to debug locally. The extensive use of triggers and bindings makes Azure Functions very pleasant to use and in many cases, removes the need to manage instances of Http clients or CosmosDB clients in the function code.

The function implements basic validations and using FluentValidator library in .NET. Having a valid payload, we create a new command and send it to CosmosDB using a very convenient output binding. With this binding, we can just call CosmosDB. We handle a possible insertion conflict by calling FlushAsync to be able to handle any exception during the insertion and possibly, return to the caller that a given request has already been processed. When the request is not valid, we use the validation messages to inform the caller about wrong or missing parameters. We expect that the payload conforms to a model that represents the payload of the data that contains all the information needed to process the command.

The API above is used to create a new credit transfer. It does not depend on external data to perform validation. In this example, we have another API that is exposed publicly, the Transactions API. We have a more complex situation, where data from the read model needs to be used to perform the validation process:

A more complex API

// TransactionsApi.cs
using FluentValidation;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.Documents;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using ServerlessStreaming.Common;
using ServerlessStreaming.DomainModel;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate.Commands;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;

namespace ServerlessStreaming.Api
{

    public class TransactionsApi
    {
        // as a client to the application, the api can use the read model to validate commands.
        private readonly ICosmosDbService<PrepaidCardAccount> prepaidAccountReadModel; 

        public TransactionsApi(ICosmosDbService<PrepaidCardAccount> prepaidAccountReadModel)
        {
            this.prepaidAccountReadModel = prepaidAccountReadModel;
        }


        [FunctionName("TransactionsApi")]
        public async Task<IActionResult> Run(

            // be sure to add content-type: application/json on your request! 
            [HttpTrigger(AuthorizationLevel.Function, "post", Route = "create")]
                HttpRequest req, 

            [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Commands",
                ConnectionStringSetting = "CosmosDBConnectionString")]
                IAsyncCollector<IncreaseCreditsCommand> increaseCreditsCommands,

            [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Commands",
                ConnectionStringSetting = "CosmosDBConnectionString")]
                IAsyncCollector<DecreaseCreditsCommand> decreaseCreditsCommands,

            ILogger log)
        {
            log.LogInformation("Starting validation of Credit Change request...");


            using var reader = new StreamReader(req.Body);
            string requestBody = await reader.ReadToEndAsync();
            var payload = JsonConvert.DeserializeObject<AccountTransactionModel>(requestBody);


            // some validation rules here: https://docs.fluentvalidation.net/en/latest/advanced.html
            var existingAccount = await prepaidAccountReadModel.GetItemAsync(payload.PrepaidCardAccountId, payload.PrepaidCardAccountId);
            var validator = new TransactionValidator();
            var validationContext = new ValidationContext<AccountTransactionModel>(payload);
            validationContext.RootContextData["account"] = existingAccount; // using the account stored on read model to perform validation of business data.
            var validationResult = validator.Validate(validationContext);

            if (!validationResult.IsValid)
            {
                return new BadRequestObjectResult(validationResult.Errors.Select(e => new {
                    Field = e.PropertyName,
                    Error = e.ErrorMessage
                }));
            }


            switch (payload.Operation)
            {
                case AccountTransactionModel.IncreaseOperation:
                {
                    return await PerformIncreaseOperation(req, payload, increaseCreditsCommands, log);
                }

                case AccountTransactionModel.DecreaseOperation:
                {
                    return await PerformDecreaseOperation(req, payload, decreaseCreditsCommands, log);
                }
                default:
                {
                    return new BadRequestObjectResult("Invalid Request: please inform operation.");
                }
            }
        }

        private static async Task<IActionResult> PerformDecreaseOperation(HttpRequest req, AccountTransactionModel creditsChangeRequest, IAsyncCollector<DecreaseCreditsCommand> decreaseCreditsCommands, ILogger log)
        {
            var command = new DecreaseCreditsCommand(aggregateId: creditsChangeRequest.PrepaidCardAccountId,
                correlationId: creditsChangeRequest.CorrelationId ?? Guid.NewGuid().ToString(),
                createdAtUtc: DateTime.UtcNow,
                payload: creditsChangeRequest);
            try
            {
                await decreaseCreditsCommands.AddAsync(command);
                await decreaseCreditsCommands.FlushAsync();

                log.LogWarning($"Decrease Credits Command generated. CorrelationId: {command.CorrelationId}, AggregateId: {command.AggregateId}");
                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                return new AcceptedResult();
            }
            catch (DocumentClientException ex) when (ex.Error.Code == "Conflict")
            {

                log.LogWarning("Returning Not Modified: ", creditsChangeRequest.Id);
                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                return new StatusCodeResult(304);
            }
        }

        private static async Task<IActionResult> PerformIncreaseOperation(HttpRequest req, AccountTransactionModel creditsChangeRequest, IAsyncCollector<IncreaseCreditsCommand> increaseCreditsCommands, ILogger log)
        {
            var command = new IncreaseCreditsCommand(aggregateId: creditsChangeRequest.PrepaidCardAccountId,
                correlationId: creditsChangeRequest.CorrelationId ?? Guid.NewGuid().ToString(),
                createdAtUtc: DateTime.UtcNow,
                payload: creditsChangeRequest);

            try
            {
                await increaseCreditsCommands.AddAsync(command);
                await increaseCreditsCommands.FlushAsync();

                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                log.LogWarning($"Increase Credits Command generated. CorrelationId: {command.CorrelationId}, AggregateId: {command.AggregateId}");
                return new AcceptedResult();
            }
            catch (DocumentClientException ex) when (ex.Error.Code == "Conflict")
            {
                log.LogWarning("Returning Not Modified: ", creditsChangeRequest.Id);
                req.HttpContext.Response.Headers.Add("ETag", command.ETag);
                return new StatusCodeResult(304);
            }
        }
    }
}

In this second example, we are demonstrating an API to increase or decrease credits for a given account. These transactions can be called from other services like the Transfers API. This function is using dependency injection to receive a dependency on a service that can do queries in Cosmos DB. This mechanism is very similar to the ASP.NET Core dependency injection infrastructure you are probably used to work with.

The validation logic is more complex. We first go to CosmosDB and look for an account. Inside the validation process, we check if the informed account owner is valid in a custom validator:

using FluentValidation;
using FluentValidation.Results;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate;

namespace ServerlessStreaming.Api
{
    public class TransactionValidator : AbstractValidator<AccountTransactionModel>
    {
        // see also: https://www.tomfaltesek.com/azure-functions-input-validation/

        public TransactionValidator()
        {
            RuleFor(x => x).NotNull();
            RuleFor(x => x.Id).NotEmpty();
            RuleFor(x => x.PrepaidCardAccountId).NotEmpty();
            RuleFor(x => x.CardholderId).NotEmpty();
            RuleFor(x => x.Amount).GreaterThan(0m);

            RuleFor(x => x.CardholderId).Custom((x, context) => {

                var account = context.ParentContext.RootContextData["account"] as PrepaidCardAccount;
                if (x != account.CardHolderId)
                {
                    context.AddFailure("The informed account does not have a valid owner.");
                }
            });
        }

        protected override bool PreValidate(ValidationContext<AccountTransactionModel> context, ValidationResult result)
        {

            if (context.InstanceToValidate == null)
            {
                result.Errors.Add(new ValidationFailure("", "Please ensure a model was supplied."));
                return false;
            }
            return true;
        }
    }
}

If there is a mismatch, we add an error to the list of validation errors and return them to the client. Another difference is that we are using two output bindings - one for each kind of transaction. The rest of the function is generally the same. We do the transformation of a request into a valid command that potentially will succeed and produce a change in the global state.

Commands and Aggregates in the world of CQRS and Event Sourcing

CQRS and Event Sourcing, as said in many other articles, work very well in combination. We will show here a complete example and we will describe how we can build our event sourced aggregates to properly organize our business logic.

Physical Organization

In our solution, we have a project to organize the domain and maintain the business rules together. In serverless architectures, it's often tempting to write business logic inside the functions. We should try to avoid this. Our domain logic should try to be host-agnostic.

The image below shows the physical structure of our example project. It's easy to see the aggregates, commands, and events.
A new developer can rapidly understand what are the main commands and events produced by the system. The structure of commands and events shown above can help to make things more intention revealing. The business rules are usually found in command handlers.

Command Structure

A command contains all the data necessary to be processed without any other dependencies. Commands are very simple structures and fairly easy to understand.

First, let's see our command implementation details:

using Newtonsoft.Json;
using System;

namespace ServerlessStreaming.Common
{

    public interface ICommand
    {
        string Id { get; set; }
        string CorrelationId { get; }
        string AggregateId { get; }
        DateTime CreatedAtUtc { get; }
        string CommandName { get; set; }
    }


    public abstract class Command<TPayload> : ICommand where TPayload : IPayload
    {
        private string commandName;

        protected Command(string aggregateId, string correlationId, DateTime createdAtUtc, TPayload payload)
        {
            CorrelationId = correlationId;
            CreatedAtUtc = createdAtUtc;
            AggregateId = aggregateId;
            Payload = payload;
            commandName = GetType().FullName;
        }


        [JsonProperty(PropertyName = "id")]
        public string Id { get; set; }

        public string CommandName
        {
            get
            {
                return commandName;
            }
            set
            {
                if (value == GetType().FullName)
                {
                    commandName = value;
                }
                else
                {
                    throw new TypeInitializationException(GetType().FullName, new Exception("The command name must match the C# full type name in order to prevent json desserialization mismatches."));
                }
            }
        }

        public string CorrelationId { get; }

        public DateTime CreatedAtUtc { get; }

        public string AggregateId { get; }

        public TPayload Payload { get; }

        public string ETag
        {
            get
            {
                //todo: needs to be opaque
                return CommandHash;
            }
        }

        public abstract string CommandHash { get; }

        public override string ToString()
        {
            try
            {
                var json = JsonConvert.SerializeObject(this);
                return json;
            }
            catch (Exception)
            {
                return $"AggregateId: {AggregateId}, CorrelationId: {CorrelationId}, Id: {Id}, CommandName: {CommandName}";
            }
        }
    }
}

The structure is not complicated. We have a Command class that requires a type parameter of type TPayload that represents the model of the command. Another extremely important field is CorrelationId. The correlation identifier is used to tie all the commands and events produced by a given request. This is mandatory for the troubleshooting of the multiple parts of the system and can help in auditing. With this unique identifier, we can query all the commands, events, and log messages on either CosmosDB or Application Insights. This is a big win for event-driven architectures. We can investigate and understand how the state was reached and we can see the motivations of each state change. A command encapsulates a motivation to change the global state. This is not true for common CRUD systems, where we only see the final state of an aggregate.

We also introduce the concept of Aggregate Id represented by another required field of the command. The aggregate id is the concrete subject of the command. The command must refer to an aggregate identified so it changes the state of this specific aggregate. The aggregate id for an account can be, for example, 1236-5649-9998-7721.

All commands are required to have a unique name and we automate this by getting the C# class name as the command name in our example. We use the command name to partition the Commands container in CosmosDB. Usually, we share the same container with multiple commands to reduce the overall solution's financial cost. Since we will have multiple commands with different JSON structures, the process of deserialization of the commands requires that we add an identifier on each command - we use the command name also to help in the deserialization process so we can find the correct deserializer. We use ETag to help the caller in client-side caching.

A concrete command should be something like this:

using ServerlessStreaming.Common;
using System;

namespace ServerlessStreaming.DomainModel.CreditsTransferAggregate.Commands
{
    public class CreateCreditsTransferCommand : Command<CreditsTransferModel>
    {
        public CreateCreditsTransferCommand(
            string aggregateId,
            string correlationId,
            DateTime createdAtUtc,
            CreditsTransferModel payload)
            : base(aggregateId, correlationId, createdAtUtc, payload) {  }


        public override string CommandHash => AggregateId + "_" + Payload.Id;
    }
}

We don't expose an empty constructor and we use CreditsTransferModel as our command payload. The type parameter of the base class informs what is the type acceptable for the model. The code for CreditsTransferModel is:

using Newtonsoft.Json;
using ServerlessStreaming.Common;
using ServerlessStreaming.DomainModel.CardholderAggregate;
using System;

namespace ServerlessStreaming.DomainModel.CreditsTransferAggregate
{
    public class CreditsTransferModel : IPayload
    {
        [JsonProperty(PropertyName = "id")]
        public string Id { get; set; }
        public CardholderInfo Sender { get; set; }
        public CardholderInfo Receiver { get; set; }
        public decimal Amount { get; set; }
        public string TransferPurpose { get; set; }
        public DateTime? DateScheduledUtc { get; set; }
        public DateTime DateCreatedUtc { get; set; }
        public DateTime? DateProcessedUtc { get; set; }
    }
}

As we can see here, the model contains the fields needed to create a valid transfer operation. We force all the models to have an Id property. The IPayload interface enforces this.

Command Handling Process: Azure Functions as a Host

As we can recall, usually commands are generated by an API and then saved on a CosmosDB Collection in our architecture. The main reason for this is to be able to react to each command using the mechanisms available on the cloud for us. So we reactively receive a valid command from CosmosDB and trigger another Azure Function. In general, every command will have a corresponding command handler. The physical structure in the code is trivial:

The command handler for CreateCreditsTransferCommand is here:

using Microsoft.Azure.Documents;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Polly;
using ServerlessStreaming.Common;
using ServerlessStreaming.DomainModel.CreditsTransferAggregate;
using ServerlessStreaming.DomainModel.CreditsTransferAggregate.Commands;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using static ServerlessStreaming.Common.UtilityFunctions;

namespace CreditsTransferProcessor
{
    public class CreateCreditsTransferCommandHandler
    {
        private readonly CreditsTransferCommandHandlers commandHandlers;

        public CreateCreditsTransferCommandHandler(CreditsTransferCommandHandlers commandHandlers)
        {
            this.commandHandlers = commandHandlers;
        }

        [FunctionName("CommandHandler_CreateCreditsTransfer")]
        public async Task Run(

            [CosmosDBTrigger(
            databaseName: "CreditsTransfers",
            collectionName: "Commands",
            ConnectionStringSetting = "CosmosDBConnectionString",
            LeaseCollectionName = "leases",
            LeaseCollectionPrefix = "CH_CCT")] // abbr of this command handler
            IReadOnlyList<Document> commandsReceived,


            [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Events",
                ConnectionStringSetting = "CosmosDBConnectionString")]
            IAsyncCollector<Event> outputEvents,


            [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Commands",
                ConnectionStringSetting = "CosmosDBConnectionString")]
            IAsyncCollector<CommandHandlerFailed> commandsFailed,

             [CosmosDB(
                databaseName: "CreditsTransfers",
                collectionName: "Commands",
                ConnectionStringSetting = "CosmosDBConnectionString")]
            IAsyncCollector<CommandHandlerSucceeded> commandsSucceeded,


            [Queue("command-handler-failed")]
            IAsyncCollector<string> deadLetterQueue,

            ILogger log)
        {

            if (commandsReceived == null)
            {
                log.LogWarning("No commands available to process!");
                return;
            }

            log.LogInformation("Number of commands to process: " + commandsReceived.Count);


            foreach (var c in commandsReceived)
            {
                CreateCreditsTransferCommand command = default;

                try
                {
                    log.LogDebug("Processing document id: {Id}...", c.Id);

                    var input = CanDesserializeAs<CreateCreditsTransferCommand>(c);
                    if (input.IsNullOrEmpty())
                    {
                        continue;
                    }

                    command = JsonConvert.DeserializeObject<CreateCreditsTransferCommand>(input);

                    var context = new Context().WithLogger(log);
                    context.Add("command", command);
                    await RetryPolicies.GetPolicy().ExecuteAsync(async (context) =>
                    {
                        if (command != null && command.CommandName == typeof(CreateCreditsTransferCommand).FullName)
                        {
                            // invoke domain model logic to decide if the command should be accepted.
                            await commandHandlers.HandleAsync(command, log, outputEvents);

                            // send command confirmation
                            var commandSucceeded = new CommandHandlerSucceeded(command.Id, command.CorrelationId, command.AggregateId, command.CommandHash);
                            await commandsSucceeded.AddAsync(commandSucceeded);

                            log.LogWarning("Command Handler succeeded: Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}",
                                                                                            command.CommandName, command.CorrelationId, command.AggregateId);
                        }
                    }, context);
                }
                catch (JsonSerializationException ex) when (ex.InnerException is TypeInitializationException)
                {
                    continue;
                }
                catch (Exception ex)
                {
                    await DeadLettering.HandleFailure(commandsFailed, deadLetterQueue, log, command, ex);
                }

                log.LogDebug("Processed document id: {Id}...", c.Id);
            }
        }
    }
}

The constructor is receiving a concrete instance of a command handler defined in our domain model. As we can see, there is a new trigger here: CosmosDBTrigger. The Azure Function runtime uses the CosmosDB Change Feed mechanism to invoke this function when a new item is added on the Commands container. There are some configurations and infrastructure to make this work but in general, we need an auxiliary collection to store some locks - the lease container - and we must make sure that we don't have a duplicated lease collection prefix.

We can also see that there are output bindings to Events collection and other bindings to store commands that are succeeded and failed. We have also a binding can store the payload of failed commands.

The function can receive more than one command and we must not lose any command. For each command received, we check if the message is useful for us. We will receive all the commands in all command handlers and we must ignore commands that we don't handle so other functions can handle it. After successful deserialization, we start a retry context and try to perform our business operation, delegating this work to the command handler that was injected in the constructor. This function works as a gateway to our domain logic. After a successful operation, the CommandHandlerSucceeded message is saved on CosmosDB.

When something fails, we do a retry. If it fails again, we first try to send a CommandHandlerFailed message to CosmosDB. If it fails again, we store the command payload on a storage queue. These are very exceptional situations and must be monitored by the operations team. The error handling logic is as follows:

using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate;
using System;
using System.Threading.Tasks;

namespace ServerlessStreaming.Common
{
    public static class DeadLettering
    {
        public static async Task HandleFailure<T>(IAsyncCollector<CommandHandlerFailed> commandsFailed, IAsyncCollector<string> commandHandlerDlq, ILogger log, Command<T> cmd, Exception ex)
            where T : IPayload
        {
            log.LogDebug("Starting command error handling using Cosmos: Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", cmd.CommandName, cmd.CorrelationId, cmd.AggregateId);

            var commandHandlerFailed = new CommandHandlerFailed(cmd.Id, cmd.CorrelationId, cmd.AggregateId, cmd.CommandHash, ex);
            try
            {
                await commandsFailed.AddAsync(commandHandlerFailed);
                log.LogError("Command Handler failed: Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}, Exception: {Exception}",
                                                    cmd.CommandName, cmd.CorrelationId, cmd.AggregateId, ex);
            }
            catch (Exception ex2)
            {
                log.LogDebug("Starting command error handling using Queue Storage after cosmos failure: Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}",
                                                   cmd.CommandName, cmd.CorrelationId, cmd.AggregateId);

                try
                {
                    await commandHandlerDlq.AddAsync(cmd.ToString());
                    await commandHandlerDlq.FlushAsync();
                    log.LogDebug("Command sent to dead letter:  Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", cmd.CommandName, cmd.CorrelationId, cmd.AggregateId);
                }
                catch (Exception ex3)
                {
                    log.LogCritical("Error sending failed command to dead letter queue. Exception: {Exception}:", ex3);
                    throw;
                }


                log.LogCritical("Command Handler and command was deadlettered: Command {CommandName}, CorrelationId: {CorrelationId}, AggregateId: {AggregateId}, Exception: {Exception}", cmd.CommandName, cmd.CorrelationId, cmd.AggregateId, ex2);

            }
        }
    }
}

As we can see above, we try our best to notify the failure to the external world. If we can use CosmosDB to communicate the command handling failure, we can react to these messages and perhaps, notify the customer using some sort of automatic notification. The storage queue is our last resort. We would have used other kinds of outputs here. The idea is to use a different persistence mechanism.

Please note again that we are using a single CosmosDB container to store all the commands and we do the same to store all the events. This is a cost optimization strategy to avoid extra costs. It would be possible to create a container for each type of command or event. This would be very analogous to a relational modeling strategy. As each new container can add cost to the solution, we generally try to avoid a bigger number of containers. The side-effect of this organization is that all the functions are called on each insert and we should ignore the messages that we don't want to handle. This kind of organization, of course, can be modified later if needed. You need to monitor the performance of your solution and find bottlenecks or cost pressure continuously.

Command Handling Process: The Command Handler

The command handler works as a small orchestrator. It receives input from the host, in our case, an Azure Function. The command handler is part of the domain logic. With a command in hands, it calls the Aggregate Root to get the command handled. It depends on an EventSourcedRepository to persist the results of command handling: Events! In an event-sourced application using CQRS and DDD concepts, we ask for the domain to process a command and produce events.

using Microsoft.Extensions.Logging;
using ServerlessStreaming.DomainModel.CreditsTransferAggregate.Commands;
using System;
using ServerlessStreaming.Common;
using ServerlessStreaming.Infrastructure;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using System.Linq;

namespace ServerlessStreaming.DomainModel.CreditsTransferAggregate
{

    public class CreditsTransferCommandHandlers 
    {
        private readonly EventSourcedRepository eventSourcedRepository;

        public CreditsTransferCommandHandlers(EventSourcedRepository eventSourcedRepository)
        {
            this.eventSourcedRepository = eventSourcedRepository;
        }


        public async Task HandleAsync(CreateCreditsTransferCommand command, ILogger hostLogger, IAsyncCollector<Event> azureOutputBinding = null)
        {
            if (azureOutputBinding != null)
            {
                eventSourcedRepository.SetWritePersistence(azureOutputBinding);
            }

            var eventStream = await eventSourcedRepository.LoadAggregateStreamAsync(command.AggregateId);

            // this is applicable only when we are creating a new aggregate. We must have no prior events.
            if (eventStream.Any())
            {
                throw new Exception("Aggregate has already been created"); // todo: better exceptions;
            }

            var creditsTransfer = new CreditsTransferAggregate(command, hostLogger); 
            creditsTransfer.DesignateCreditsTransferHandlingStrategy(command);

            await eventSourcedRepository.SaveToAggregateStreamAsync(creditsTransfer.Changes);

        }


        public async Task HandleAsync(ManuallyApproveCreditsTransferCommand command, ILogger hostLogger, IAsyncCollector<Event> azureOutputBinding = null)
        {
            if (azureOutputBinding != null)
            {
                eventSourcedRepository.SetWritePersistence(azureOutputBinding);
            }

            var eventStream = await eventSourcedRepository.LoadAggregateStreamAsync(command.AggregateId);
            var creditsTransfer = new CreditsTransferAggregate(eventStream, hostLogger);
            creditsTransfer.PerformManualApproval(command);

            await eventSourcedRepository.SaveToAggregateStreamAsync(creditsTransfer.Changes);
        }



        public async Task HandleAsync(ManuallyRejectCreditsTransferCommand command, ILogger azureFunctionsLogger, IAsyncCollector<Event> azureOutputBinding = null)
        {
            if (azureOutputBinding != null)
            {
                eventSourcedRepository.SetWritePersistence(azureOutputBinding);
            }

            var eventStream = await eventSourcedRepository.LoadAggregateStreamAsync(command.AggregateId);
            var creditsTransfer = new CreditsTransferAggregate(eventStream, azureFunctionsLogger);
            creditsTransfer.PerformManualRejection(command);

            await eventSourcedRepository.SaveToAggregateStreamAsync(creditsTransfer.Changes);
        }
    }
}

As we can see in this code, the command handler knows how to call the aggregate. It calls a public method on the aggregate root and captures the Changes property to persist the events to the event-sourced repository. Since we are using the output binding mechanism of the host, Azure Functions, we need to slightly change the command handler to receive this infrastructure dependency and configure the repository to understand IAsyncCollector<T>. We are also using the logging infrastructure of Azure in the form of a common ILogger.

Understanding the Event Stream

Before going into the demonstration of how an aggregate root is implemented, it's important to visualize a concrete example of what an event sourced aggregate means. It's far easier to explain using an example.

Consider that a new account is created. Then some credits are added. The account number created is 1234-4321-5678-0987. The aggregate name is Account.

Event Name	Aggregate Version	Event Body	Balance (current state of this aggregate)
Account Created	0	`{AggregateId: 1234-4321-5678-0987}`	$ 0.00
Credits Increased	1	`{AggregateId: 1234-4321-5678-0987, Amount: 1000.00}`	$ 1000.00
Credits Decreased	2	`{AggregateId: 1234-4321-5678-0987, Amount: 10.00}`	$ 990.00

As we can see, we only save to the database these events. We don't perform an update. Never. A database that only appends events is called event store in the industry. We are using CosmosDB as our event store. As you can imagine, the current state of Account aggregate is derived from the event stream. We must read the entire event stream to calculate the current state of the aggregate. This is a major mindset change. The repository code and aggregate code now need to understand how to work with an event stream. We admit that this is not trivial and requires some study time, but you should consider this strategy to create new systems that are not small. Using event sourcing is an advantage over traditional database-state based development because all the changes in the global state are documented in the event stream. The global state is the entire event stream. We can easily understand how a given aggregate evolved and we gain the ability to do time-travel debugging to identify and solve issues.

Aggregate Versions and Concurrency

Suppose there are two instances of the DecreaseCreditsCommandHandler that intend to decrease $500.00, almost at the same time (race condition). This would be possible if we just added events to the stream without checking anything, allowing the user to have then a negative balance. The situation would be like this:

Event Name	Aggregate Version	Event Body	Balance (current state of this aggregate)
Account Created	0	`{AggregateId: 1234-4321-5678-0987}`	$ 0.00
Credits Increased	1	`{AggregateId: 1234-4321-5678-0987, Amount: 1000.00}`	$ 1000.00
Credits Decreased	2	`{AggregateId: 1234-4321-5678-0987, Amount: 10.00}`	$ 990.00
Credits Decreased	3	`{AggregateId: 1234-4321-5678-0987, Amount: 500.00}`	$ 490.00
Credits Decreased	3	`{AggregateId: 1234-4321-5678-0987, Amount: 500.00}`	$ -10.00

We can see that we are breaking a domain invariant that avoids accounts with a negative amount of credits. To avoid this problem we should reject the command that would lead to an invalid state. A solution to this problem would be to query the event stream and detect the current aggregate version. We then try to publish a new event incrementing the version. If there is an event with the same version, we do a retry a little bit later.

Command Name	Event Name	Aggregate Version	Event Body	Balance (current state of this aggregate)
CreateNewAccount	Account Created	0	`{AggregateId: 1234-4321-5678-0987}`	$ 0.00
IncreaseCredits	Credits Increased	1	`{AggregateId: 1234-4321-5678-0987, Amount: 1000.00}`	$ 1000.00
DecreaseCredits	Credits Decreased	2	`{AggregateId: 1234-4321-5678-0987, Amount: 10.00}`	$ 990.00
DecreaseCredits thread#1	Credits Decreased	3	`{AggregateId: 1234-4321-5678-0987, Amount: 500.00}`	$ 490.00
DecreaseCredits thread#2	-	still 3 because there is another event with version 3	no event published	$ 490.00
DecreaseCredits thread#2 (retry)	Credits Decrease Rejected	4	`{AggregateId: 1234-4321-5678-0987, Amount: 500.00, Reason: "Insufficient Funds"}`	$ 490.00

Thread #2 tries to decrease the credits and fails. Tries again and command succeeds, now, producing an event explaining that there were invalid funds.

We have added a uniqueness constraint on the Events collection so we can get an exception upon the insertion of a duplicated version for an aggregate.

Command Handlers and Read Models

As we saw in the previous post, the read model is eventually consistent. It means that an event published will eventually be reflected in the read model. Usually, the read model update interval is very slow and is imperceptible to the users. Some developers can be tempted to access the read model to perform some additional validations and run business logic. This is not correct. There is always a risk of looking at stale data. The command handler should only look at the event stream of the aggregate it intends to affect. Deviations of this rule should be carefully discussed. Access to read models affected by other aggregates can be considered, but usually, the UI of the application should feed the command with data from diverse read models. The API that accepts requests from the UI can also do this kind of preparation work to issue correct commands, eventually, obtaining data from multiple read models.

An Event Sourced Repository

Our event-sourced repository is little bit unconventional:

using Microsoft.Azure.WebJobs;
using Newtonsoft.Json;
using ServerlessStreaming.Common;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;


namespace ServerlessStreaming.Infrastructure
{
    public class EventSourcedRepository 
    {
        private readonly ICosmosDBEventStore<Event> eventStore;
        private IAsyncCollector<Event> azureOutputBinding;

        public EventSourcedRepository(ICosmosDBEventStore<Event> eventStore)
        {
            this.eventStore = eventStore;
        }


        public async Task SaveToAggregateStreamAsync(IEnumerable<Event> events)
        {
            if (azureOutputBinding == null)
            {
                await SaveToAggregateStreamUsingEventStore(events);
            }
            else
            {
                await SaveToAggregateStreamUsingAzureOutputBinding(events, azureOutputBinding);
            }
        }

        private async Task SaveToAggregateStreamUsingAzureOutputBinding(IEnumerable<Event> events, IAsyncCollector<Event> azureOutputBinding)
        {

            foreach (var item in events)
            {
                await azureOutputBinding.AddAsync(item);
            }
            await azureOutputBinding.FlushAsync();
        }

        private async Task SaveToAggregateStreamUsingEventStore(IEnumerable<Event> events)
        {
            // todo: optization & cache - not performant.
            foreach (var e in events)
            {
                var concreteEventType = AppDomain.CurrentDomain.GetAssemblies().Select(x => x.GetType(e.EventName)).Where(x => x != null).First();
                var concreteEvent = JsonConvert.DeserializeObject(e.ToString(), concreteEventType);
                await eventStore.AddItemRawAsync(concreteEvent, e.EventName);
            }
        }

        public async Task<IEnumerable<Event>> LoadAggregateStreamAsync(string aggregateId)
        {

            var eventStoreQuery = $@" SELECT c.id, c.EventName, c.EventDateTimeUtc, c.AggregateId, c.AggregateVersion, c.CorrelationId, c.Payload 
                                      FROM c
                                      WHERE c.AggregateId = '{aggregateId}' 
                                      ORDER BY c.AggregateVersion";

            // we get all the events still as Cosmos DB documents. 
            var events = await eventStore.GetItemsRawAsync(eventStoreQuery);

            var concreteEvents = new List<Event>();

            foreach (var ev in events)
            {
                var concreteEventTypeName = ev.GetPropertyValue<string>("EventName");
                var concreteEventType = AppDomain.CurrentDomain.GetAssemblies()
                                                               .Select(x => x.GetType(concreteEventTypeName))
                                                               .Where(x => x != null).First();

                if (JsonConvert.DeserializeObject(ev.ToString(), concreteEventType) is Event concreteEvent)
                {
                    concreteEvents.Add(concreteEvent);
                }
                else
                {
                    throw new InvalidOperationException("Could not process event or the event is unknown.");
                }
            }

            return concreteEvents;
        }

        public EventSourcedRepository SetWritePersistence(IAsyncCollector<Event> outputBinding)
        {
            this.azureOutputBinding = outputBinding;
            return this;
        }
    }
}

This repository works adding events to an aggregate's stream and loading the stream back when needed. The repository depends on a low-level service to interact with CosmosDB. We also can work adding items to either our CosmosDB service or a IAsyncCollector<Event>.

We must be aware that most of the domain logic is placed on the write side of the application. The changes in the global state are controlled by the aggregates, command handlers, and specialized repositories that work with the event store. We should not be tempted to query directly the event store to get data, for example, to satisfy UI requests. Since the read and write sides of the application are physically separated, as CQRS recommends, we use the read side to query the read models and then, build the UI. This repository is used only to manage the events in the event store.

Understanding the Aggregate Root

Let's go back to the command handling logic. What we do is briefly is:

Receive a command in an Azure Function that is triggered by the insertion of a new command in CosmosDB
Call the Command Handler
The Command Handler calls the Aggregate
The Aggregate works with the command payload and produces events
The Command Handler persists those events using the event-sourced repository.

We now need to understand the details of an Aggregate Root and how it effectively handles business logic. Let's start with the implementation of the base AggregateRoot class:

using Microsoft.Extensions.Logging;
using ServerlessStreaming.Common;
using System;
using System.Collections.Generic;
using System.Linq;

namespace ServerlessStreaming.DomainModel
{
    public abstract class AggregateRoot
    {
        protected AggregateRoot(IEnumerable<Event> eventStream, ILogger log)
        {
            this.Log = log;
            if (eventStream.Any() == false)
            {
                throw new Exception("Could not load aggregate from the event stream because it is empty.");
            }

            this.EventStream.AddRange(eventStream);
            LoadFromHistory(this.EventStream);
            LoadAggregateIdFromStream();

            Log.LogDebug("Aggregate State Restored.");
        }

        protected AggregateRoot(ILogger log) {
            this.Log = log;
        }

        private string aggregateId;

        public string AggregateId
        {
            get => aggregateId;

            protected set
            {
                if (value.IsNullOrEmpty())
                {
                    throw new ArgumentException("The Aggregate Id must not be empty.");
                }
                aggregateId = value;
            }
        }
        protected List<Event> EventStream { get; } = new List<Event>();

        public List<Event> Changes { get; } = new List<Event>();

        protected ILogger Log { get; }

        public abstract void Apply(Event evt);

        protected abstract void LoadFromHistory(IEnumerable<Event> eventStream);

        protected virtual void LoadAggregateIdFromStream()
        {
            this.AggregateId = EventStream.First().AggregateId;
        }
    }
}

The Aggregate Root is responsible to manage a group of related objects so they can work together as a unit to the external world. When we want to reconstruct the aggregate, we pass an event stream to the constructor so we can process each event ordered by the aggregate version. Sometimes we need to create a new aggregate, like a new user account, without any event stream. So we have two constructors available. The logging is required in our project.

During reconstruction, we process each event and update the state of the aggregate. This is done when LoadFromHistory is called:

 protected override void LoadFromHistory(IEnumerable<Event> eventStream)
 {
     foreach (var evt in eventStream)
     {
         _ = evt switch
         {
             CreditsDecreasedEvent e => Apply(e, false),
             CreditsIncreasedEvent e => Apply(e, false),

             CreditsDecreaseRejectedEvent e => Apply(e, false),
             CreditsIncreaseRejectedEvent e => Apply(e, false),

             CreditsDecreaseReversedEvent e => Apply(e, false),
             CreditsIncreaseReversedEvent e => Apply(e, false),
         };

         currentAggregateVersion = evt.AggregateVersion;
     }
 }

We have to Apply each event received. Like this:

 private int Apply(CreditsDecreasedEvent e, bool isNew)
 {
     state.CurrentBalance -= (e.Payload as AccountTransactionModel).Amount;
     state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;

     if (isNew)
     {
         Changes.Add(e);
     }

     return e.AggregateVersion;
 }

We are applying each preexistent event and updating the local state of the Aggregate. With an updated state, the aggregate is capable of doing correct decisions and generate correct events. Please note that the current aggregate version is loaded during the LoadAggregateIdFromStream call. The aggregate id is also obtained from any event on the stream. We use the first one. Please note that each event that changes the state must have a corresponding Apply method.

Please note that this reconstruction of the state is something that only occurs in the memory of the host running the code of the Aggregate Root. The calculated state is usually not saved anywhere unless we do a process called "snapshot", which is a photography of the state ate a given moment that is also saved in the stream. A snapshot can be a performance optimization when the event stream is very long. Please also note that we can work on the raw events processing in any way we want - we can do different projections of the events in the event store for different purposes. This is pure gold. You can read the stream and use it for different use cases, potentially unknown yet.

When the reconstruction of the aggregate is completed, the command handler is ready to dispatch commands to the aggregate:

 public void DecreaseCredits(DecreaseCreditsCommand command)
 {
     var currentBalance = state.CurrentBalance;

     // check if we have enough credits to be accept this command:
     var amountToDecrease = command.Payload.Amount;
     if (amountToDecrease <= state.CurrentBalance)
     {
         // if so, we can do the debit.
         ProceedWithCreditsDecrease(command, currentBalance, currentAggregateVersion);
     }
     else
     {
         // if not, we have to communicate the rejection using an Event
         RejectCreditsDecrease(command, currentBalance, currentAggregateVersion);
     }
 }


 private void ProceedWithCreditsDecrease(DecreaseCreditsCommand command, decimal currentBalance, int currentAggregateVersion)
 {
     var ev = new CreditsDecreasedEvent(
         command.CorrelationId,
         DateTime.UtcNow,
         command.AggregateId,
         currentAggregateVersion + 1,
         command.Payload)
     {
         PreviousBalance = currentBalance,
         CurrentBalance = currentBalance - command.Payload.Amount

     };

     Log.LogWarning("Credits Decrease will be accepted. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);

     Apply(ev);
 }

The aggregate is simply exposing the DecreaseCredits method and deciding the command should be accepted or rejected. When the credits decrease is accepted, we create CreditsDecreasedEvent, incrementing the aggregate version and filling the necessary fields. We also log this to increase the visibility of the operations even during development time. The event can also contain custom fields. We tie the previous and current balance in the event. This is useful to troubleshoot the balance variation easily. And probably, this makes it more difficult to tamper the balance.

We have to communicate the existence of this new event to the world. We have to put all the new events on the Changes list. The Apply method again is called:

 public override void Apply(Event evt)
 {
     var aggregateVersion = evt switch
     {
         CreditsDecreasedEvent e => Apply(e, true),
         CreditsIncreasedEvent e => Apply(e, true),

         CreditsDecreaseRejectedEvent e => Apply(e, true),
         CreditsIncreaseRejectedEvent e => Apply(e, true),

         CreditsDecreaseReversedEvent e => Apply(e, true),
         CreditsIncreaseReversedEvent e => Apply(e, true),
     };

     currentAggregateVersion = aggregateVersion;
 }

 private int Apply(CreditsDecreasedEvent e, bool isNew)
 {
     state.CurrentBalance -= (e.Payload as AccountTransactionModel).Amount;
     state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;

     if (isNew)
     {
         Changes.Add(e);
     }

     return e.AggregateVersion;

 }

We apply the new event and update again the local state, as other calls to this aggregate can occur. The Changes collection is used by the command handler logic to send the events to the repository:

   eventSourcedRepository.SaveToAggregateStreamAsync(prepaidCardAccountAggregate.Changes);

The state of the aggregate can be any object. The PrepaidCardAccountAggregate uses the entity PrepaidCardAccount to store the current state of the aggregate.

 private readonly PrepaidCardAccount state = new PrepaidCardAccount();
 private int currentAggregateVersion = -1;

 public PrepaidCardAccountAggregate(IEnumerable<Event> eventStream, ILogger log)
  : base(eventStream, log) { }

// rest of the PrepaidCardAccountAggregate ommitted

An entity is just a simple class or a class that can collaborate with other entities and value objects. The code for PrepaidCardAccount entity is as follows:


using Newtonsoft.Json;
using System;

namespace ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate
{
    public class PrepaidCardAccount
    {
        [JsonProperty(PropertyName = "id")]
        public string Id { get; set; } 
        public string CardHolderId { get; set; }
        public decimal CurrentBalance { get; set; }
        public DateTime AccountCreatedAtUtc { get; set; }
        public DateTime BalanceUpdatedAtUtc { get; set; }
        public int AggregateVersion { get; set; }
    }
}

Now we can see the whole code for PrepaidCardAccountAggregate. Please forgive us, but we felt that for C# and .NET there are only a few complete examples of how to implement a realistic aggregate root.

using Microsoft.Extensions.Logging;
using ServerlessStreaming.Common;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate.Commands;
using ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace ServerlessStreaming.DomainModel.PrepaidCardAccountAggregate
{
    public class PrepaidCardAccountAggregate : AggregateRoot
    {

        private readonly PrepaidCardAccount state = new PrepaidCardAccount();
        private int currentAggregateVersion = -1;


        public PrepaidCardAccountAggregate(IEnumerable<Event> eventStream, ILogger log)
         : base(eventStream, log) { }


        public void DecreaseCredits(DecreaseCreditsCommand command)
        {
            var currentBalance = state.CurrentBalance;

            // check if we have enough credits to be accept this command:
            var amountToDecrease = command.Payload.Amount;
            if (amountToDecrease <= state.CurrentBalance)
            {
                // if so, we can do the debit.
                ProceedWithCreditsDecrease(command, currentBalance, currentAggregateVersion);
            }
            else
            {
                // if not, we have to communicate the rejection using an Event
                RejectCreditsDecrease(command, currentBalance, currentAggregateVersion);
            }
        }


        public void IncreaseCredits(IncreaseCreditsCommand command)
        {

            var currentBalance = state.CurrentBalance;

            var ev = new CreditsIncreasedEvent(command.CorrelationId, DateTime.UtcNow, command.AggregateId, currentAggregateVersion + 1, command.Payload)
            {
                PreviousBalance = currentBalance,
                CurrentBalance = currentBalance + command.Payload.Amount
            };

            Log.LogWarning("Credits Increase will be accepted. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);

            Apply(ev);

        }

        public void ReverseCreditsChange(ReverseCreditsChangeCommand command)
        {
            var eventToRevert = EventStream.Single(x => (x.Payload as IPayload).Id == command.Payload.PayloadId);

            if (eventToRevert.EventName.Contains(nameof(CreditsIncreasedEvent)))
            {
                ReverseCreditsIncrease(command, eventToRevert as CreditsIncreasedEvent);
            }
            else
            {
                ReverseCreditsDecrease(command, eventToRevert as CreditsDecreasedEvent);
            }

        }

        private void ReverseCreditsIncrease(ReverseCreditsChangeCommand command, CreditsIncreasedEvent eventToRevert)
        {
            var currentBalance = state.CurrentBalance;
            var amountToRevert = (eventToRevert.Payload as AccountTransactionModel).Amount;

            var evt = new CreditsIncreaseReversedEvent(command.CorrelationId,
                                                       DateTime.UtcNow,
                                                       command.AggregateId,
                                                       currentAggregateVersion + 1,
                                                       command.Payload)
            {
                PreviousBalance = currentBalance,
                AmountToRevert = amountToRevert,
                CurrentBalance = currentBalance - amountToRevert
            };

            Log.LogWarning("Credits Increase will be reversed. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);

            Apply(evt);
        }


        private void ReverseCreditsDecrease(ReverseCreditsChangeCommand command, CreditsDecreasedEvent eventToRevert)
        {
            var currentBalance = state.CurrentBalance;
            var amountToRevert = (eventToRevert.Payload as AccountTransactionModel).Amount;

            var evt = new CreditsDecreaseReversedEvent(command.CorrelationId,
                                                       DateTime.UtcNow,
                                                       command.AggregateId,
                                                       currentAggregateVersion + 1,
                                                       command.Payload)
            {
                PreviousBalance = currentBalance,
                AmountToRevert = amountToRevert,
                CurrentBalance = currentBalance + amountToRevert
            };

            Log.LogWarning("Credits Decrease will be reversed. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);

            Apply(evt);
        }

        private void ProceedWithCreditsDecrease(DecreaseCreditsCommand command, decimal currentBalance, int currentAggregateVersion)
        {
            var ev = new CreditsDecreasedEvent(
                command.CorrelationId,
                DateTime.UtcNow,
                command.AggregateId,
                currentAggregateVersion + 1,
                command.Payload)
            {
                PreviousBalance = currentBalance,
                CurrentBalance = currentBalance - command.Payload.Amount

            };

            Log.LogWarning("Credits Decrease will be accepted. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);


            Apply(ev);
        }

        private void RejectCreditsDecrease(DecreaseCreditsCommand command, decimal currentBalance, int currentAggregateVersion)
        {
            var ev = new CreditsDecreaseRejectedEvent(
                command.CorrelationId,
                DateTime.UtcNow,
                command.AggregateId,
                currentAggregateVersion + 1,
                command.Payload)
            {
                PreviousBalance = currentBalance,
                AmountRequested = command.Payload.Amount,
                Reason = "Insufficient Credits. Requested amount: " + command.Payload.Amount
            };

            Log.LogWarning("Credits Decrease will be rejected due to insufficiente funds. CorrelationId: {CorrelationId}, AggregateId: {AggregateId}", command.CorrelationId, AggregateId);

            Apply(ev);
        }


        public override void Apply(Event evt)
        {
            var aggregateVersion = evt switch
            {
                CreditsDecreasedEvent e => Apply(e, true),
                CreditsIncreasedEvent e => Apply(e, true),

                CreditsDecreaseRejectedEvent e => Apply(e, true),
                CreditsIncreaseRejectedEvent e => Apply(e, true),

                CreditsDecreaseReversedEvent e => Apply(e, true),
                CreditsIncreaseReversedEvent e => Apply(e, true),
            };

            currentAggregateVersion = aggregateVersion;
        }

        protected override void LoadFromHistory(IEnumerable<Event> eventStream)
        {
            foreach (var evt in eventStream)
            {
                _ = evt switch
                {
                    CreditsDecreasedEvent e => Apply(e, false),
                    CreditsIncreasedEvent e => Apply(e, false),

                    CreditsDecreaseRejectedEvent e => Apply(e, false),
                    CreditsIncreaseRejectedEvent e => Apply(e, false),

                    CreditsDecreaseReversedEvent e => Apply(e, false),
                    CreditsIncreaseReversedEvent e => Apply(e, false),
                };

                currentAggregateVersion = evt.AggregateVersion;
            }
        }



        private int Apply(CreditsDecreasedEvent e, bool isNew)
        {
            state.CurrentBalance -= (e.Payload as AccountTransactionModel).Amount;
            state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;

            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;

        }

        private int Apply(CreditsIncreasedEvent e, bool isNew)
        {
            state.CurrentBalance += (e.Payload as AccountTransactionModel).Amount;
            state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;

            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;
        }

        private int Apply(CreditsDecreaseRejectedEvent e, bool isNew)
        {
            // no state changes right now. But this can change in the future.

            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;
        }

        private int Apply(CreditsIncreaseRejectedEvent e, bool isNew)
        {
            // no state changes right now. But this can change in the future.
            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;
        }

        private int Apply(CreditsDecreaseReversedEvent e, bool isNew) 
        {

            state.CurrentBalance += e.AmountToRevert;
            state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;


            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;
        }

        private int Apply(CreditsIncreaseReversedEvent e, bool isNew)
        {
            state.CurrentBalance -= e.AmountToRevert;
            state.BalanceUpdatedAtUtc = e.EventDateTimeUtc;

            if (isNew)
            {
                Changes.Add(e);
            }

            return e.AggregateVersion;
        }
    }
}

The advantages of this design, despite the need to change the mindset, is a more testable aggregate root and the clarity of what the system can do, given certain inputs. This kind of code is very testable and the business logic is not coupled to the host. The use of either Azure Functions or other .NET host is irrelevant when we centralize the business rules in a cohesive domain model. It is possible to simulate many situations and validate if the domain invariants are still preserved.

Conclusion of Part 2

Wow! We were able to describe how the commands are processed by a simple cloud structure using serverless components: Azure Functions and CosmosDB. This article is an effort to explain concretely many concepts that are spread around many blogs and documentation on the web, giving some north for developers that are planning to create applications on Azure in a more structured way still keeping the joy of serverless technologies in terms of flexibility and costs. The use of DDD and CQRS/ES concepts are not restricted to monoliths or services running on-premises or on Kubernetes.

Having the right mindset it is viable to CQRS, DDD, Event Sourcing in a reactive and serverless environment. Azure is a great place

In part 3 we will describe how events are effectively published and handled.