Add media download service (part 1)

package.json

@@ -12,6 +12,8 @@
 		"output-handler::start": "AWS_REGION=eu-west-1 STAGE=DEV npm run start --workspace output-handler",
 		"worker-capacity-manager::build": "npm run build --workspace worker-capacity-manager",
 		"worker-capacity-manager::start": "AWS_REGION=eu-west-1 STAGE=DEV npm run start --workspace worker-capacity-manager",
+		"media-download::build": "npm run build --workspace media-download",
+		"media-download::start": "AWS_REGION=eu-west-1 STAGE=DEV npm run start --workspace media-download",
 		"worker::build": "npm run build --workspace worker; npm run build --workspace worker",
 		"worker::package": "npm run package --workspace worker",
 		"worker::start": "AWS_REGION=eu-west-1 STAGE=DEV npm run start --workspace worker",

packages/api/src/index.ts

@@ -19,12 +19,15 @@ import {
 	logger,
 	getObjectText,
 	getS3Client,
+	sendMessage,
 } from '@guardian/transcription-service-backend-common';
 import {
 	ClientConfig,
 	TranscriptExportRequest,
 	inputBucketObjectMetadata,
 	transcribeFileRequestBody,
+	transcribeUrlRequestBody,
+	MediaDownloadJob,
 } from '@guardian/transcription-service-common';
 import type { SignedUrlResponseBody } from '@guardian/transcription-service-common';
 import {
@@ -71,6 +74,43 @@ const getApp = async () => {
 		}),
 	]);
 
+	apiRouter.post('/transcribe-url', [
+		checkAuth,
+		asyncHandler(async (req, res) => {
+			const userEmail = req.user?.email;
+			const body = transcribeUrlRequestBody.safeParse(req.body);
+			const id = uuid4();
+			if (!body.success || !userEmail) {
+				res.status(422).send('missing request params');
+				return;
+			}
+			const downloadJob: MediaDownloadJob = {
+				id,
+				url: body.data.url,
+				languageCode: body.data.languageCode,
+				translationRequested: body.data.translationRequested,
+				userEmail,
+			};
+
+			const sendResult = await sendMessage(
+				sqsClient,
+				config.app.mediaDownloadQueueUrl,
+				JSON.stringify(downloadJob),
+				id,
+			);
+			if (isSqsFailure(sendResult)) {
+				res.status(500).send(sendResult.errorMsg);
+				return;
+			}
+			logger.info('API successfully sent the message to SQS', {
+				id,
+				url: body.data.url,
+				userEmail,
+			});
+			res.send('Message sent');
+		}),
+	]);
+
 	apiRouter.post('/transcribe-file', [
 		checkAuth,
 		asyncHandler(async (req, res) => {
@@ -125,35 +165,12 @@ const getApp = async () => {
 				body.data.fileName,
 				signedUrl,
 				body.data.languageCode,
-				false,
+				body.data.translationRequested,
 			);
 			if (isSqsFailure(sendResult)) {
 				res.status(500).send(sendResult.errorMsg);
 				return;
 			}
-			if (body.data.translationRequested) {
-				const translationSendResult =
-					await generateOutputSignedUrlAndSendMessage(
-						s3Key,
-						sqsClient,
-						config.app.taskQueueUrl,
-						config.app.transcriptionOutputBucket,
-						config.aws.region,
-						userEmail,
-						body.data.fileName,
-						signedUrl,
-						body.data.languageCode,
-						true,
-					);
-				if (isSqsFailure(translationSendResult)) {
-					res
-						.status(500)
-						.send(
-							`Translation request failed: ${translationSendResult.errorMsg}`,
-						);
-					return;
-				}
-			}
 			logger.info('API successfully sent the message to SQS', {
 				id: s3Key,
 				filename: body.data.fileName,

packages/backend-common/package.json

@@ -6,7 +6,7 @@
 		"@aws-sdk/client-s3": "^3.624.0",
 		"@aws-sdk/client-sqs": "^3.624.0",
 		"@aws-sdk/client-ssm": "^3.624.0",
-		"@aws-sdk/lib-dynamodb": "^3.624.0",
+		"@aws-sdk/lib-dynamodb": "3.624.0",
 		"@aws-sdk/client-cloudwatch": "^3.624.0",
 		"@aws-sdk/client-auto-scaling": "^3.624.0",
 		"axios": "^1.7.4",

packages/backend-common/src/config.ts

@@ -13,6 +13,7 @@ export interface TranscriptionConfig {
 		rootUrl: string;
 		taskQueueUrl: string;
 		deadLetterQueueUrl?: string;
+		mediaDownloadQueueUrl: string;
 		stage: string;
 		emailNotificationFromAddress: string;
 		sourceMediaBucket: string;
@@ -76,6 +77,11 @@ export const getConfig = async (): Promise<TranscriptionConfig> => {
 
 	logger.info(`Parameters fetched: ${parameterNames.join(', ')}`);
 	const taskQueueUrl = findParameter(parameters, paramPath, 'taskQueueUrl');
+	const mediaDownloadQueueUrl = findParameter(
+		parameters,
+		paramPath,
+		'mediaDownloadQueueUrl',
+	);
 	const deadLetterQueueUrl =
 		stage === 'DEV'
 			? undefined
@@ -139,6 +145,7 @@ export const getConfig = async (): Promise<TranscriptionConfig> => {
 			secret: appSecret,
 			taskQueueUrl,
 			deadLetterQueueUrl,
+			mediaDownloadQueueUrl,
 			stage,
 			sourceMediaBucket,
 			emailNotificationFromAddress,

packages/backend-common/src/process.ts

@@ -0,0 +1,55 @@
+import { spawn } from 'child_process';
+import { logger } from './logging';
+export interface ProcessResult {
+	code?: number;
+	stdout: string;
+	stderr: string;
+}
+
+export const runSpawnCommand = (
+	processName: string,
+	cmd: string,
+	args: ReadonlyArray<string>,
+	logStdout: boolean,
+): Promise<ProcessResult> => {
+	return new Promise((resolve, reject) => {
+		const cp = spawn(cmd, args);
+		const stdout: string[] = [];
+		const stderr: string[] = [];
+		cp.stdout.on('data', (data) => {
+			stdout.push(data.toString());
+		});
+
+		cp.stderr.on('data', (data) => {
+			stderr.push(data.toString());
+		});
+
+		cp.on('error', (e) => {
+			stderr.push(e.toString());
+		});
+
+		cp.on('close', (code) => {
+			const result = {
+				stdout: stdout.join(''),
+				stderr: stderr.join(''),
+				code: code || undefined,
+			};
+			if (logStdout) {
+				logger.info(
+					`process ${processName} stdout: ${result.stdout} stderr: ${result.stderr}`,
+				);
+			} else {
+				logger.info('Logging stderr only');
+				logger.info(`process ${processName} stderr: ${result.stderr}`);
+			}
+			if (code === 0) {
+				resolve(result);
+			} else {
+				logger.error(
+					`process ${processName} failed with code ${result.code} due to: ${result.stderr}`,
+				);
+				reject(result);
+			}
+		});
+	});
+};

packages/backend-common/src/sqs.ts

@@ -66,18 +66,18 @@ export const generateOutputSignedUrlAndSendMessage = async (
 	originalFilename: string,
 	inputSignedUrl: string,
 	languageCode: LanguageCode,
-	translate: boolean,
+	translationRequested: boolean,
 ): Promise<SendResult> => {
 	const signedUrls = await generateOutputSignedUrls(
 		s3Key,
 		region,
 		outputBucket,
 		userEmail,
 		7,
-		translate,
+		translationRequested,
 	);
 
-	const jobId = translate ? `${s3Key}-translation` : s3Key;
+	const jobId = translationRequested ? `${s3Key}-translation` : s3Key;
 	const job: TranscriptionJob = {
 		id: jobId, // id of the source file
 		inputSignedUrl,
@@ -87,12 +87,32 @@ export const generateOutputSignedUrlAndSendMessage = async (
 		originalFilename,
 		outputBucketUrls: signedUrls,
 		languageCode,
-		translate,
+		translate: false,
 	};
-	return await sendMessage(client, queueUrl, JSON.stringify(job), s3Key);
+	const messageResult = await sendMessage(
+		client,
+		queueUrl,
+		JSON.stringify(job),
+		s3Key,
+	);
+	if (isSqsFailure(messageResult) && translationRequested) {
+		logger.info(
+			`Failed to send message, error message: ${messageResult.errorMsg}`,
+		);
+		return messageResult;
+	}
+	if (!isSqsFailure(messageResult) && translationRequested) {
+		return await sendMessage(
+			client,
+			queueUrl,
+			JSON.stringify({ ...job, translate: true }),
+			s3Key,
+		);
+	}
+	return messageResult;
 };
 
-const sendMessage = async (
+export const sendMessage = async (
 	client: SQSClient,
 	queueUrl: string,
 	messageBody: string,