postgresql - 处理来自 kinesis 的事件时出现奇怪的问题

Question

我在 aws 上设置了 amazon connect，如果我进行测试调用，它会将该调用放入 aws kinesis 流中。我正在尝试编写一个处理这些记录并将它们保存到数据库的 lambda。如果我打一个简单的电话（拨打号码 - asnwer - hangup），它工作得很好。但是，如果我进行多部分呼叫（呼叫一个号码 - 接听 - 转移到另一个号码 - 挂断），这会作为两个单独的记录 (CTR) 出现在 kinesis 中。我的 lambda 会一一处理 CTR（联系跟踪记录）。首先，它将 CTR 保存到名为 call_segments 的表中，然后查询该表以查看该调用的其他部分是否已经存在。如果是，则合并数据并保存到名为 completed_calls 的表中，否则跳过它。如果一个呼叫有多个段（如果它被转移到另一个号码），它会将它作为两个事件带给您。

这是我的代码：

const callRecordService = require("./call-records-service");
exports.handler = async (event) => {
await Promise.all(
    event.Records.map(async (record) => {
        return processRecord(record);
    })
);
};
const processRecord = async function(record)        {
    try{
        const payloadStr = new Buffer(record.kinesis.data, "base64").toString("ascii");
        let payload = JSON.parse(payloadStr);
    await     callRecordService.processCTR(payload);
}
catch(err){
    // console.error(err);
}
};

这是服务文件：

async function processCTR(ctr) {
let userId = "12"
let result = await saveCtrToContactDetails(ctr, userId);
    let paramsForCallSegments = [ctr.InstanceARN.split("instance/").pop(), ctr.ContactId]
    let currentCallSegements = await dbHelper.getAll(dbQueries.getAllCallSegmentsQuery, paramsForCallSegments)
    let completedCall = checkIfCallIsComplete(currentCallSegements);
    if (completedCall) {
        console.log('call is complete')
       let results = await saveCallToCompletedCalls(completedCall);
    }
}

//------------- Private functions --------------------
const saveCtrToContactDetails = async (ctr, userId) => {
let params = [ctr.ContactId,userId,ctr.callDuration];
let results = await dbHelper.executeQuery(dbQueries.getInsertCallDetailsRecordsQuery, params);
return results;
}

const checkIfCallIsComplete = (currentCallSegements) => {
   //This function checks if all callSegments are already in call_segments table.

}

const saveCallToCompletedCalls = async (completedCall) => {
    let contact_id = completedCall[0].contact_id;
    let user_id = completedCall[0].user_id;
    let call_duration = completedCall[0] + completedCall[1]
    completedCall.forEach(callSegment => {
        call_duration += callSegment.call_duration;
});
let params = [contact_id, user_id, call_duration];
let results = await dbHelper.executeQuery(dbQueries.getInsertToCompletedCallQuery, params);
};

标签： postgresqlaws-lambdaamazon-kinesisamazon-connect