Indexing S3 Images Atomically

cdk.ts

import * as cdk from "@aws-cdk/core";
import * as lambda from "@aws-cdk/aws-lambda";
import * as iam from "@aws-cdk/aws-iam";
import * as sqs from "@aws-cdk/aws-sqs";
import * as s3n from "@aws-cdk/aws-s3-notifications";
import * as s3 from "@aws-cdk/aws-s3";

const queue = new sqs.Queue();
const fn = new lambda.Function();
const bucket = new s3.Bucket();

fn.grantInvoke(new iam.ServicePrincipal("sqs.amazonaws.com"));

fn.addEventSourceMapping(`${functionName}${queueName}SqsEventSource`, {
  batchSize: 100,
  maxBatchingWindow: cdk.Duration.minutes(5),
  eventSourceArn: queue.queueArn,
  retryAttempts: 10
});

queue.grantConsumeMessages(fn);

const destination = new s3n.SqsDestination(queue);
bucket.addEventNotification(eventType, destination);

query.sql

INSERT INTO device_camera_frames (s3_bucket, s3_key, s3_record_status, s3_event_time, controller_id, frame_timestamp, camera_id)
VALUES ("bucket", "key", 'deleted', 1, "controller1", 1, 1), ("bucket", "key", 'exists', 0, "controller1", 1, 1) 
ON DUPLICATE KEY UPDATE
s3_record_status = IF(s3_event_time < VALUES(s3_event_time), VALUES(s3_record_status), s3_record_status),
s3_event_time = IF(s3_event_time < VALUES(s3_event_time), VALUES(s3_event_time), s3_event_time);

table.sql

CREATE TABLE device_camera_frames (
    s3_bucket varchar(255),
    s3_key varchar(255),
    s3_record_status varchar(10) NOT NULL,
    s3_event_time BIGINT UNSIGNED NOT NULL,
    controller_id varchar(255) NOT NULL,
    frame_timestamp BIGINT UNSIGNED NOT NULL,
    camera_id INT NOT NULL,
    PRIMARY KEY (s3_bucket, s3_key)
);