Implemented events adapter

packages/framework-provider-azure-infrastructure/src/infrastructure/stacks/application-stack.ts

@@ -25,6 +25,19 @@ export class ApplicationStackBuilder {
       {},
       '../templates/storage-account.json'
     )
+    const cosmosDbDeployment = await this.buildResource(
+      resourceManagementClient,
+      resourceGroupName,
+      {
+        databaseName: { value: this.config.resourceNames.applicationStack },
+        containerName: { value: this.config.resourceNames.eventsStore },
+        partitionKey: { value: eventStorePartitionKeyAttribute },
+      },
+      '../templates/cosmos-db-account.json'
+    )
+
+    const cosmosDbConnectionString = cosmosDbDeployment.properties?.outputs.connectionString.value
+
     const functionAppDeployment = await this.buildResource(
       resourceManagementClient,
       resourceGroupName,
@@ -82,6 +95,8 @@ export class ApplicationStackBuilder {
     appSettings.properties.BOOSTER_ENV = this.config.environmentName
     // @ts-ignore
     appSettings.properties.BOOSTER_REST_API_URL = `https://${apiManagementServiceDeployment.properties?.outputs.apiManagementServiceName.value}.azure-api.net/${this.config.environmentName}`
+    // @ts-ignore
+    appSettings.properties.COSMOSDB_CONNECTION_STRING = cosmosDbConnectionString
 
     // update app settings
     await webSiteManagementClient.webApps.updateApplicationSettings(
@@ -92,24 +107,48 @@ export class ApplicationStackBuilder {
       }
     )
 
-    const zipPath = await this.packageAzureFunction('graphql', {
-      bindings: [
-        {
-          authLevel: 'anonymous',
-          type: 'httpTrigger',
-          direction: 'in',
-          name: 'rawRequest',
-          methods: ['post'],
+    const zipPath = await this.packageAzureFunction([
+      {
+        functionName: 'graphql',
+        functionConfig: {
+          bindings: [
+            {
+              authLevel: 'anonymous',
+              type: 'httpTrigger',
+              direction: 'in',
+              name: 'rawRequest',
+              methods: ['post'],
+            },
+            {
+              type: 'http',
+              direction: 'out',
+              name: '$return',
+            },
+          ],
+          scriptFile: '../dist/index.js',
+          entryPoint: this.config.serveGraphQLHandler.split('.')[1],
         },
-        {
-          type: 'http',
-          direction: 'out',
-          name: '$return',
+      },
+      {
+        functionName: 'eventHandler',
+        functionConfig: {
+          bindings: [
+            {
+              type: 'cosmosDBTrigger',
+              name: 'documents',
+              direction: 'in',
+              leaseCollectionName: 'leases',
+              connectionStringSetting: 'COSMOSDB_CONNECTION_STRING',
+              databaseName: this.config.resourceNames.applicationStack,
+              collectionName: this.config.resourceNames.eventsStore,
+              createLeaseCollectionIfNotExists: 'true',
+            },
+          ],
+          scriptFile: '../dist/index.js',
+          entryPoint: this.config.eventDispatcherHandler.split('.')[1],
         },
-      ],
-      scriptFile: '../dist/index.js',
-      entryPoint: 'boosterServeGraphQL',
-    })
+      },
+    ])
 
     // @ts-ignore
     const deployResponse = await this.deployFunctionPackage(
@@ -169,15 +208,21 @@ export class ApplicationStackBuilder {
     )
   }
 
-  private async packageAzureFunction(functionName: string, functionConfig: object): Promise<any> {
+  private async packageAzureFunction(
+    functionDefinitions: Array<{ functionName: string; functionConfig: object }>
+  ): Promise<any> {
     const output = fs.createWriteStream(os.tmpdir() + '/example.zip')
     const archive = archiver('zip', {
       zlib: { level: 9 }, // Sets the compression level.
     })
 
     archive.pipe(output)
     archive.glob('**/*')
-    archive.append(JSON.stringify(functionConfig, null, 2), { name: functionName + '/function.json' })
+    functionDefinitions.forEach((functionDefinition) => {
+      archive.append(JSON.stringify(functionDefinition.functionConfig, null, 2), {
+        name: functionDefinition.functionName + '/function.json',
+      })
+    })
     archive.finalize()
 
     return new Promise((resolve, reject) => {

packages/framework-provider-azure-infrastructure/src/infrastructure/templates/cosmos-db-account.json

@@ -21,7 +21,10 @@
     }
   },
   "functions": [],
-  "variables": {},
+  "variables": {
+    "resourceId": "[resourceId('Microsoft.DocumentDB/databaseAccounts', parameters('cosmosDbAccountName'))]",
+    "apiVersion": "[providers('Microsoft.DocumentDB', 'databaseAccounts').apiVersions[0]]"
+  },
   "resources": [
     {
       "name": "[parameters('cosmosDbAccountName')]",
@@ -94,5 +97,18 @@
       ]
     }
   ],
-  "outputs": {}
+  "outputs": {
+    "documentEndpoint": {
+      "type": "string",
+      "value": "[reference(variables('resourceId'), variables('apiVersion')).documentEndpoint]"
+    },
+    "accountKey": {
+      "type": "string",
+      "value": "[listKeys(variables('resourceId'), variables('apiVersion')).primaryMasterKey]"
+    },
+    "connectionString": {
+      "type": "string",
+      "value": "[concat('AccountEndpoint=https://', parameters('cosmosDbAccountName'), '.documents.azure.com:443/;AccountKey=', listKeys(variables('resourceId'), variables('apiVersion')).primaryMasterKey, ';')]"
+    }
+  }
 }

packages/framework-provider-azure/package.json

@@ -20,6 +20,7 @@
     "url": "git+https://github.com/boostercloud/booster.git"
   },
   "dependencies": {
+    "@azure/cosmos": "^3.7.2",
     "@azure/functions": "^1.2.2",
     "@boostercloud/framework-types": "^0.3.3"
   },

packages/framework-provider-azure/src/constants.ts

@@ -6,4 +6,5 @@ export const subscriptionsStoreTTLAttribute = 'expirationTime'
 export const environmentVarNames = {
   restAPIURL: 'BOOSTER_REST_API_URL',
   websocketAPIURL: 'BOOSTER_WEBSOCKET_API_URL',
+  cosmosDbConnectionString: 'COSMOSDB_CONNECTION_STRING',
 } as const

packages/framework-provider-azure/src/index.ts

@@ -2,14 +2,30 @@
 import { ProviderInfrastructure, ProviderLibrary } from '@boostercloud/framework-types'
 import { requestFailed, requestSucceeded } from './library/api-adapter'
 import { rawGraphQLRequestToEnvelope } from './library/graphql-adapter'
+import {
+  rawEventsToEnvelopes,
+  storeEvents,
+  readEntityEventsSince,
+  readEntityLatestSnapshot,
+} from './library/events-adapter'
+import { CosmosClient } from '@azure/cosmos'
+import { environmentVarNames } from './constants'
+
+let cosmosClient
+if (process.env[environmentVarNames.cosmosDbConnectionString]) {
+  // @ts-ignore
+  cosmosClient = new CosmosClient(process.env[environmentVarNames.cosmosDbConnectionString])
+} else {
+  cosmosClient = undefined
+}
 
 export const Provider: ProviderLibrary = {
   // ProviderEventsLibrary
   events: {
-    rawToEnvelopes: undefined as any,
-    store: undefined as any,
-    forEntitySince: undefined as any,
-    latestEntitySnapshot: undefined as any,
+    rawToEnvelopes: rawEventsToEnvelopes,
+    store: storeEvents.bind(null, cosmosClient),
+    forEntitySince: readEntityEventsSince.bind(null, cosmosClient),
+    latestEntitySnapshot: readEntityLatestSnapshot.bind(null, cosmosClient),
   },
   // ProviderReadModelsLibrary
   readModels: {

packages/framework-provider-azure/src/library/events-adapter.ts

@@ -0,0 +1,107 @@
+import { EventEnvelope } from '@boostercloud/framework-types'
+import { Container, CosmosClient, Database, ItemResponse } from '@azure/cosmos'
+import { BoosterConfig, Logger, UUID } from '@boostercloud/framework-types'
+import { eventStorePartitionKeyAttribute, eventStoreSortKeyAttribute } from '../constants'
+import { partitionKeyForEvent } from './partition-keys'
+
+// eslint-disable-next-line @typescript-eslint/no-magic-numbers
+const originOfTime = new Date(0).toISOString()
+
+export function rawEventsToEnvelopes(rawEvents: Array<any>): Array<EventEnvelope> {
+  return rawEvents.map(
+    (rawEvent: any): EventEnvelope => {
+      return rawEvent as EventEnvelope
+    }
+  )
+}
+
+export async function readEntityEventsSince(
+  cosmosDb: CosmosClient,
+  config: BoosterConfig,
+  logger: Logger,
+  entityTypeName: string,
+  entityID: UUID,
+  since?: string
+): Promise<Array<EventEnvelope>> {
+  const fromTime = since ? since : originOfTime
+  const database: Database = cosmosDb.database(config.resourceNames.applicationStack)
+  const container: Container = database.container(config.resourceNames.eventsStore)
+  const { resources } = await container.items
+    .query({
+      query: `SELECT * FROM c where c.${eventStorePartitionKeyAttribute} = @partitionKey AND c.${eventStoreSortKeyAttribute} > @fromTime ORDER BY c.${eventStoreSortKeyAttribute} DESC`,
+      parameters: [
+        {
+          name: '@partitionKey',
+          value: partitionKeyForEvent(entityTypeName, entityID),
+        },
+        {
+          name: '@fromTime',
+          value: fromTime,
+        },
+      ],
+    })
+    .fetchAll()
+  return resources.map((resource: any) => {
+    return resource as EventEnvelope
+  })
+}
+
+export async function readEntityLatestSnapshot(
+  cosmosDb: CosmosClient,
+  config: BoosterConfig,
+  logger: Logger,
+  entityTypeName: string,
+  entityID: UUID
+): Promise<EventEnvelope | null> {
+  const database: Database = cosmosDb.database(config.resourceNames.applicationStack)
+  const container: Container = database.container(config.resourceNames.eventsStore)
+  const { resources } = await container.items
+    .query({
+      query: `SELECT * FROM c where c.${eventStorePartitionKeyAttribute} = @partitionKey ORDER BY c.${eventStoreSortKeyAttribute} DESC LIMIT 1`,
+      parameters: [
+        {
+          name: '@partitionKey',
+          value: partitionKeyForEvent(entityTypeName, entityID, 'snapshot'),
+        },
+      ],
+    })
+    .fetchAll()
+
+  const snapshot = resources[0]
+  if (snapshot) {
+    logger.debug(
+      `[EventsAdapter#readEntityLatestSnapshot] Snapshot found for entity ${entityTypeName} with ID ${entityID}:`,
+      snapshot
+    )
+    return snapshot as EventEnvelope
+  } else {
+    logger.debug(
+      `[EventsAdapter#readEntityLatestSnapshot] No snapshot found for entity ${entityTypeName} with ID ${entityID}.`
+    )
+    return null
+  }
+}
+
+export async function storeEvents(
+  cosmosDb: CosmosClient,
+  eventEnvelopes: Array<EventEnvelope>,
+  config: BoosterConfig,
+  logger: Logger
+): Promise<void> {
+  logger.debug('[EventsAdapter#storeEvents] Storing EventEnvelopes with eventEnvelopes:', eventEnvelopes)
+  const database: Database = cosmosDb.database(config.resourceNames.applicationStack)
+  const container: Container = database.container(config.resourceNames.eventsStore)
+  const events: Array<Promise<ItemResponse<any>>> = eventEnvelopes.map((eventEnvelope) => {
+    return container.items.create({
+      ...eventEnvelope,
+      [eventStorePartitionKeyAttribute]: partitionKeyForEvent(
+        eventEnvelope.entityTypeName,
+        eventEnvelope.entityID,
+        eventEnvelope.kind
+      ),
+      [eventStoreSortKeyAttribute]: new Date().toISOString(),
+    })
+  })
+  await Promise.all(events)
+  logger.debug('[EventsAdapter#storeEvents] EventEnvelope stored')
+}

packages/framework-provider-azure/src/library/partition-keys.ts

@@ -0,0 +1,9 @@
+import { EventEnvelope, UUID } from '@boostercloud/framework-types'
+
+export function partitionKeyForEvent(
+  entityTypeName: string,
+  entityID: UUID,
+  kind: EventEnvelope['kind'] = 'event'
+): string {
+  return `${entityTypeName}-${entityID}-${kind}`
+}