fix(lunchbot): added table for storing restaurants and visits into them

Author: JussiLem

.projen/deps.json

@@ -19,17 +19,18 @@ const project = new awscdk.AwsCdkTypeScriptApp({
   },
   projenrcTs: true,
   deps: [
+    '@types/aws-lambda',
     '@aws-lambda-powertools/logger',
     '@aws-lambda-powertools/metrics',
     '@aws-lambda-powertools/tracer',
     '@middy/core',
     '@aws-sdk/client-lex-runtime-v2',
     '@aws-sdk/client-dynamodb',
     '@aws-sdk/lib-dynamodb',
+    '@aws-sdk/util-dynamodb',
   ],
   devDeps: [
     'eslint-plugin-functional@6.6.3',
-    '@types/aws-lambda',
     'commitizen',
     'cz-conventional-changelog',
     'semantic-release',

.projen/tasks.json

@@ -0,0 +1,67 @@
+import { AttributeValue } from '@aws-sdk/client-dynamodb'
+import { UpdateCommand, UpdateCommandInput } from '@aws-sdk/lib-dynamodb'
+import { unmarshall } from '@aws-sdk/util-dynamodb'
+import { DynamoDBStreamHandler } from 'aws-lambda'
+import { CustomSlot } from './customSlot'
+import { Lunch } from './lunchState'
+import { dbClient } from '../common/dbClient'
+import { logger } from '../common/powertools'
+import { ensureError } from '../ensureError'
+
+/**
+ * Increase visit by one for given restaurant
+ * @param lunches
+ */
+const updateVisitsToDynamoDb = async (lunches: Lunch[]): Promise<void> => {
+  // Map each lunch for a promise that updates the visits count
+  const updatePromises = lunches.map((lunch) => {
+    logger.debug('Lunch state', { lunch })
+
+    const restaurant = lunch.restaurant
+    const officeLocation = lunch.officeLocation
+    if (!restaurant) {
+      return Promise.reject(new Error('Restaurant not found'))
+    }
+
+    const input: UpdateCommandInput = {
+      TableName: process.env.RESTAURANT_TABLE!,
+      Key: {
+        restaurant,
+        officeLocation,
+      },
+      UpdateExpression: 'ADD #visits :inc',
+      ExpressionAttributeNames: {
+        '#visits': 'visits',
+      },
+      ExpressionAttributeValues: {
+        ':inc': 1,
+      },
+    }
+
+    logger.debug('Updating state', { input })
+    return dbClient.send(new UpdateCommand(input))
+  })
+
+  try {
+    // Wait for all update operations to complete
+    await Promise.all(updatePromises)
+  } catch (e) {
+    const error = ensureError(e)
+    // Handle errors here, potentially logging them or taking other actions
+    logger.error('Error updating visits to DynamoDB:', error)
+  }
+}
+export const handler: DynamoDBStreamHandler = async (event) => {
+  logger.debug('Received event', { event })
+  const records: Lunch[] = event.Records.flatMap((record) => {
+    return record.dynamodb?.NewImage
+      ? unmarshall(
+          record.dynamodb.NewImage as { [key: string]: AttributeValue },
+        )
+      : []
+  })
+  const lunches = records.filter(
+    (record) => record.slot === CustomSlot.Restaurants,
+  )
+  await updateVisitsToDynamoDb(lunches)
+}

.projenrc.ts

@@ -0,0 +1,16 @@
+import { CustomSlot } from './customSlot'
+
+export interface LunchState {
+  sessionId: string
+  slot: CustomSlot
+  slotValue: { [key: string]: string }
+  expireAt: number
+}
+
+export interface Lunch {
+  sessionId: string
+  slot: CustomSlot
+  restaurant?: string
+  officeLocation?: string
+  expireAt: number
+}

package-lock.json

@@ -231,7 +231,10 @@ export const processSlots: NextSlotHandler = async (
       await storeState({
         sessionId,
         slot: CustomSlot.Restaurants,
-        slotValue: { restaurant: extractedRestaurant },
+        slotValue: {
+          restaurant: extractedRestaurant,
+          officeLocation: officeLocation,
+        },
         expireAt: Math.floor(Date.now() / 1000) + ONE_WEEK_IN_SECONDS,
       })
       return createCloseAction(sessionAttributes, intent, [

package.json

@@ -1,33 +1,28 @@
 import { PutCommand, PutCommandInput } from '@aws-sdk/lib-dynamodb'
-import { CustomSlot } from './customSlot'
 import { dbClient } from '../common/dbClient'
 import { logger } from '../common/powertools'
 import { ensureError } from '../ensureError'
+import { LunchState } from './lunchState'
 
 const stateTableName = process.env.STATE_TABLE!
 
-interface LunchState {
-  sessionId: string
-  slot: CustomSlot
-  slotValue: { [key: string]: string }
-  expireAt: number
-}
-
 /**
  * If the item already exists (i.e., an item with the same id and intentName)
  * @return null
  * @param sessionId
- * @param intentName
+ * @param slot
  * @param slotValue
+ * @param expireAt
  * @example
- * // Example usage with a dynamic slotValue key
+ * // Example usage with multiple slotValues
  * const sessionId = 'session123';
- * const slot = { name: 'lunchSlot', type: 'type1' };
- * const slotValue = { myDynamicKey: 'myDynamicValue' };
+ * const slot = CustomSlot.LunchSlot;
+ * const slotValue = { restaurant: 'Place1', officeLocation: 'Location1' };
  * await storeState({
  * sessionId,
  * slot,
- * slotValue
+ * slotValue,
+ * expireAt: Math.floor(Date.now() / 1000) + ONE_WEEK_IN_SECONDS
  * });
  */
 export const storeState = async ({
@@ -36,19 +31,23 @@ export const storeState = async ({
   slotValue,
   expireAt,
 }: LunchState): Promise<void | null> => {
-  const [key, value] = Object.entries(slotValue)[0]
+  // Using object spread to build the item
+  const item: PutCommandInput['Item'] = {
+    id: sessionId,
+    slot,
+    expireAt,
+    ...slotValue,
+  }
+
   const input: PutCommandInput = {
     TableName: stateTableName,
-    Item: {
-      id: sessionId,
-      slot: slot,
-      [key]: value,
-      expireAt,
-    },
+    Item: item,
     ConditionExpression:
-      'attribute_not_exists(id) AND attribute_not_exists(intentName)',
+      'attribute_not_exists(id) AND attribute_not_exists(slot)',
   }
+
   logger.debug('Updating state', { input })
+
   try {
     await dbClient.send(new PutCommand(input))
     return // Return if the new item was created successfully

src/fulfillment/lunch-stream.ts

@@ -1,14 +1,14 @@
 import {
-  aws_glue as glue,
+  Aws,
   aws_dynamodb as dynamodb,
+  aws_glue as glue,
+  aws_iam as iam,
   aws_lambda as lambda,
+  aws_lambda_event_sources as event_sources,
   aws_lambda_nodejs as nodejs,
-  aws_iam as iam,
+  aws_logs as logs,
   RemovalPolicy,
-  Aws,
-  // Aws,
 } from 'aws-cdk-lib'
-import { AttributeType } from 'aws-cdk-lib/aws-dynamodb'
 import { Construct } from 'constructs'
 
 export class Lunch extends Construct {
@@ -44,12 +44,13 @@ export class Lunch extends Construct {
         name: 'id',
       },
       sortKey: {
-        type: AttributeType.STRING,
+        type: dynamodb.AttributeType.STRING,
         name: 'slot',
       },
       deletionProtection: false,
       timeToLiveAttribute: 'expireAt',
       removalPolicy: RemovalPolicy.DESTROY,
+      dynamoStream: dynamodb.StreamViewType.NEW_IMAGE,
     })
     const crawlerRole = new iam.Role(this, 'CrawlerRole', {
       assumedBy: new iam.ServicePrincipal('glue.amazonaws.com'),
@@ -116,6 +117,34 @@ export class Lunch extends Construct {
         entry: 'src/fulfillment/fulfillment.ts',
       },
     )
+    const restaurantTable = new dynamodb.TableV2(this, 'Restaurant', {
+      partitionKey: {
+        type: dynamodb.AttributeType.STRING,
+        name: 'restaurant',
+      },
+      sortKey: {
+        type: dynamodb.AttributeType.STRING,
+        name: 'officeLocation',
+      },
+      removalPolicy: RemovalPolicy.DESTROY,
+    })
+    const streamLambda = new nodejs.NodejsFunction(this, 'streamLambda', {
+      runtime: lambda.Runtime.NODEJS_20_X,
+      logRetention: logs.RetentionDays.ONE_MONTH,
+      environment: {
+        SERVICE_NAME: 'lunchbot',
+        POWERTOOLS_LOG_LEVEL: 'ERROR',
+        RESTAURANT_TABLE: restaurantTable.tableName,
+      },
+      entry: 'src/fulfillment/lunch-stream.ts',
+    })
+    streamLambda.addEventSource(
+      new event_sources.DynamoEventSource(stateTable, {
+        startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+      }),
+    )
+    restaurantTable.grantReadWriteData(streamLambda)
+    stateTable.grantStreamRead(streamLambda)
     stateTable.grantReadWriteData(this.fulfillmentLambda)
     lunchTable.grantReadData(this.fulfillmentLambda)
   }