| 5 | import { queue } from "@console/core/util/queue"; |
| 6 | |
| 7 | export const handler = async (event: KinesisStreamEvent) => |
| 8 | withActor( |
| 9 | { |
| 10 | type: "public", |
| 11 | properties: {}, |
| 12 | }, |
| 13 | async () => { |
| 14 | console.log("got", event.Records.length, "records"); |
| 15 | const incomplete: string[] = event.Records.map( |
| 16 | (r) => r.eventID, |
| 17 | ).reverse(); |
| 18 | await queue(5, event.Records, async (record) => { |
| 19 | console.log( |
| 20 | "arrival", |
| 21 | new Date( |
| 22 | record.kinesis.approximateArrivalTimestamp * 1000, |
| 23 | ).toISOString(), |
| 24 | new Date().toISOString(), |
| 25 | "diff", |
| 26 | Date.now() - record.kinesis.approximateArrivalTimestamp * 1000, |
| 27 | ); |
| 28 | if ( |
| 29 | Date.now() - record.kinesis.approximateArrivalTimestamp * 1000 > |
| 30 | 1000 * 60 * 5 |
| 31 | ) { |
| 32 | incomplete.pop(); |
| 33 | console.log("too old"); |
| 34 | return; |
| 35 | } |
| 36 | const decoded = JSON.parse( |
| 37 | unzipSync(Buffer.from(record.kinesis.data, "base64")).toString(), |
| 38 | ); |
| 39 | if (decoded.messageType !== "DATA_MESSAGE") { |
| 40 | incomplete.pop(); |
| 41 | return; |
| 42 | } |
| 43 | try { |
| 44 | await Issue.extract(decoded); |
| 45 | incomplete.pop(); |
| 46 | } catch (ex) { |
| 47 | console.error(ex); |
| 48 | } |
| 49 | }); |
| 50 | |
| 51 | console.log("incomplete", incomplete.length); |
| 52 | const response = { |
| 53 | batchItemFailures: incomplete.map((id) => ({ |
| 54 | itemIdentifier: id, |
| 55 | })), |
| 56 | }; |
| 57 | |
| 58 | return response; |
| 59 | }, |
| 60 | ); |