Amazon Connectの通話記録をKinesis Data Streamsに配信時、Step Functionsでデータの重複対策を実装する

{
  "Comment": "通話記録内容をBacklogチケットに起票する",
  "StartAt": "Base64Decode",
  "States": {
    "Base64Decode": {
      "Type": "Pass",
      "Next": "ChannelTypeCheck",
      "Assign": {
        "decodedData": "{% $base64decode($states.input.data) %}"
      }
    },
    "ChannelTypeCheck": {
      "Type": "Choice",
      "Choices": [
        {
          "Next": "RecordContactInDynamoDB",
          "Condition": "{% $parse($decodedData).Channel = \"VOICE\" %}",
          "Assign": {
            "data": "{% $parse($decodedData) %}"
          }
        }
      ],
      "Default": "NonVoiceChannelEnd"
    },
    "RecordContactInDynamoDB": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Arguments": {
        "TableName": "cm-hirai-connect-backlog-tracker",
        "Item": {
          "contact_id": {
            "S": "{% $data.ContactId %}"
          },
          "ttl": {
            "N": "{% $string($floor($toMillis($now()) / 1000 + 86400)) %}"
          }
        },
        "ConditionExpression": "attribute_not_exists(contact_id)"
      },
      "Next": "InquiryTypeCheck",
      "Catch": [
        {
          "ErrorEquals": [
            "DynamoDB.ConditionalCheckFailedException"
          ],
          "Next": "DuplicateContactEnd",
          "Comment": "既に同じコンタクトIDが存在する場合は重複処理と判断して終了する"
        },
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "Next": "Fail",
          "Comment": "その他のエラーが発生した場合は処理を失敗させる"
        }
      ]
    },
    "DuplicateContactEnd": {
      "Type": "Succeed",
      "Comment": "既に処理済みのコンタクトIDなので処理を終了"
    },
    "InquiryTypeCheck": {
      "Type": "Choice",
      "Choices": [
        {
          "Next": "ExtractContactData",
          "Condition": "{% $data.Attributes.inquiry_type = \"product\" %}",
          "Assign": {
            "categoryId": "<categoryId_product>"
          }
        },
        {
          "Next": "ExtractContactData",
          "Assign": {
            "categoryId": "<categoryId_contract>"
          },
          "Condition": "{% $data.Attributes.inquiry_type = \"contract\" %}"
        }
      ],
      "Default": "ExtractContactData",
      "Assign": {
        "categoryId": "<categoryId_other>"
      }
    },
    "NonVoiceChannelEnd": {
      "Type": "Succeed"
    },
    "ExtractContactData": {
      "Type": "Pass",
      "Assign": {
        "InitialContactId": "{% $data.ContactId %}",
        "InitiationTimestamp": "{% $data.InitiationTimestamp %}",
        "DisconnectTimestamp": "{% $data.DisconnectTimestamp %}",
        "AgentName": "{% $data.Agent.Username %}",
        "CallerPhoneNumber": "{% $data.SystemEndpoint.Address %}"
      },
      "Next": "FormatContactData"
    },
    "FormatContactData": {
      "Type": "Pass",
      "Assign": {
        "FormattedInitiationTimestamp": "{% $fromMillis($toMillis($InitiationTimestamp), '[Y0001]年[M]月[D]日 [H]時[m01]分[s01]秒', '+0900') %}",
        "FormattedDisconnectTimestamp": "{% $fromMillis($toMillis($DisconnectTimestamp), '[Y0001]年[M]月[D]日 [H]時[m01]分[s01]秒', '+0900') %}",
        "FormattedPhoneNumber": "{% $substring($CallerPhoneNumber, 0, 3) = '+81' ? '0' & $substring($CallerPhoneNumber, 3) : $CallerPhoneNumber %}",
        "ContactDetailsURL": "{% 'https://<Connectドメイン>/connect/contact-trace-records/details/' & $InitialContactId & '?tz=Asia/Tokyo' %}"
      },
      "Next": "CreateBacklogTicket"
    },
    "CreateBacklogTicket": {
      "Type": "Task",
      "Resource": "arn:aws:states:::http:invoke",
      "Arguments": {
        "ApiEndpoint": "https://<xxx.backlog.jp>/api/v2/issues",
        "InvocationConfig": {
          "ConnectionArn": "<ConnectionArn>"
        },
        "Headers": {
          "Content-Type": "application/x-www-form-urlencoded"
        },
        "Transform": {
          "RequestBodyEncoding": "URL_ENCODED",
          "RequestEncodingOptions": {
            "ArrayFormat": "INDICES"
          }
        },
        "Method": "POST",
        "RequestBody": {
          "projectId": "<projectId>",
          "summary": "通話記録",
          "description": "{% '|項目|内容|備考|\n|---|---|---|\n|通話開始時間|' & $FormattedInitiationTimestamp & '| |\n|通話終了時間|' & $FormattedDisconnectTimestamp & '| |\n|コンタクトID|' & $InitialContactId & '| [コンタクト詳細URL](' & $ContactDetailsURL & ')|\n|対応者|' & $AgentName & '| |\n|発信者電話番号|' & $FormattedPhoneNumber & '| |' %}",
          "issueTypeId": "<issueTypeId>",
          "categoryId[]": "{% $categoryId %}",
          "priorityId": "3"
        }
      },
      "Next": "UpdateDynamoDBWithBacklogTicketId",
      "Retry": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "BackoffRate": 2,
          "IntervalSeconds": 1,
          "MaxAttempts": 3,
          "JitterStrategy": "FULL"
        }
      ],
      "Assign": {
        "keyId": "{% $string($states.result.ResponseBody.keyId) %}"
      }
    },
    "UpdateDynamoDBWithBacklogTicketId": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:updateItem",
      "Arguments": {
        "TableName": "cm-hirai-connect-backlog-tracker",
        "Key": {
          "contact_id": {
            "S": "{% $data.ContactId %}"
          }
        },
        "UpdateExpression": "SET backlog_ticket_id = :keyId",
        "ExpressionAttributeValues": {
          ":keyId": {
            "N": "{% $keyId %}"
          }
        }
      },
      "End": true,
      "Catch": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "Next": "Fail"
        }
      ]
    },
    "Fail": {
      "Type": "Fail"
    }
  },
  "QueryLanguage": "JSONata"
}