Lambda stores incoming messages in s3

ingestion-lambda/package.json

@@ -2,12 +2,14 @@
 	"name": "ingestion-lambda",
 	"version": "1.0.0",
 	"description": "",
-	"devDependencies": {},
 	"scripts": {
 		"dev": "tsx watch src/handler.ts",
 		"typecheck": "tsc -noEmit",
 		"build": "esbuild src/handler.ts --bundle --minify --outfile=dist/handler.js --external:aws-sdk --platform=node",
 		"test": "jest --detectOpenHandles --config ../jest.config.js --selectProjects ingestion-lambda"
 	},
-	"dependencies": {}
+	"dependencies": {
+		"@aws-sdk/client-s3": "3.637.0",
+		"@types/aws-lambda": "8.10.145"
+	}
 }

ingestion-lambda/src/config.ts

@@ -0,0 +1,20 @@
+/**
+ * Is this application running locally, or in AWS?
+ * LAMBDA_TASK_ROOT & AWS_EXECUTION_ENV are set when running in AWS
+ * See: https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html
+ */
+export const isRunningLocally =
+	!process.env.LAMBDA_TASK_ROOT && !process.env.AWS_EXECUTION_ENV;
+
+// We use localstack to mock AWS services if we are running locally.
+export const awsOptions = isRunningLocally
+	? {
+			endpoint: 'http://localhost:4566',
+			region: 'eu-west-1',
+			forcePathStyle: true,
+			credentials: {
+				accessKeyId: '',
+				secretAccessKey: '',
+			},
+		}
+	: {};

ingestion-lambda/src/handler.ts

@@ -1,7 +1,63 @@
-export const main = () => {
-	console.log('hello, world');
+import { PutObjectCommand } from '@aws-sdk/client-s3';
+import type { SNSMessage, SQSBatchResponse, SQSEvent } from 'aws-lambda';
+import { s3Client } from './s3';
+
+const BUCKET_NAME = process.env.FEEDS_BUCKET_NAME ?? 'local-feeds-bucket';
+
+export const main = async (event: SQSEvent): Promise<SQSBatchResponse> => {
+	const eventBodies = event.Records.map(
+		(record) => JSON.parse(record.body) as unknown as SNSMessage,
+	);
+	const responses = await Promise.all(
+		eventBodies.map(({ Message, MessageAttributes, MessageId }) => {
+			const fingerpostMessageId = MessageAttributes['Message-Id']?.Value;
+
+			if (!fingerpostMessageId) {
+				return Promise.resolve({
+					status: 'rejected',
+					MessageId,
+				});
+			}
+
+			const resp = s3Client.send(
+				new PutObjectCommand({
+					Bucket: BUCKET_NAME,
+					Key: `${fingerpostMessageId}.json`,
+					Body: JSON.stringify(Message),
+				}),
+			);
+			return resp
+				.then(() => ({ status: 'resolved', MessageId }))
+				.catch(() => ({ status: 'rejected', MessageId }));
+		}),
+	);
+
+	const batchItemFailures = responses
+		.filter(({ status }) => status === 'rejected')
+		.map(({ MessageId }) => ({ itemIdentifier: MessageId }));
+
+	return { batchItemFailures };
 };
 
 if (require.main === module) {
-	void (() => main())();
+	const dummyEvent: SQSEvent = {
+		Records: [
+			{
+				body: JSON.stringify({
+					Message: { feedContent: 'hello world' },
+					MessageAttributes: { 'Message-Id': { Value: '123' } },
+					MessageId: 'abc',
+				}),
+			},
+			{
+				body: JSON.stringify({
+					Message: { feedContent: 'we expect this record to fail' },
+					MessageAttributes: {},
+					MessageId: 'def',
+				}),
+			},
+		],
+	} as SQSEvent;
+
+	void (async () => console.log(await main(dummyEvent)))();
 }

ingestion-lambda/src/s3.ts

@@ -0,0 +1,19 @@
+import { S3Client } from '@aws-sdk/client-s3';
+import { awsOptions } from './config';
+
+export const s3Client = new S3Client(awsOptions);
+
+/**
+ * The location of an object in S3
+ *
+ * Provides an abstraction over the S3 bucket and object key,
+ * so we don't have to reach into the AWS SDK to get the values in tests,
+ * and prevents us from having to use the capitalized property names.
+ *
+ * @param bucketName The name of the bucket
+ * @param objectKey The key of the object
+ **/
+export type ObjectLocation = {
+	bucketName: string;
+	objectKey: string;
+};