feat: Improved S3 client (#169)

matvp91 · web-flow · commit 689f70ced29d · 2025-02-19T17:05:53.000+01:00
* Dropped s3-sync-client with custom code

* Removed s3-sync-client as dependency
diff --git a/apps/artisan/package.json b/apps/artisan/package.json
@@ -22,7 +22,6 @@
     "iso-language-codes": "^2.0.0",
     "mime-types": "^2.1.35",
     "parse-filepath": "^1.0.2",
-    "s3-sync-client": "^4.3.1",
     "shared": "workspace:*",
     "zod": "^3.24.2"
   },
diff --git a/apps/artisan/src/lib/s3.ts b/apps/artisan/src/lib/s3.ts
@@ -1,15 +1,23 @@
-import { createReadStream } from "node:fs";
-import { GetObjectCommand, S3 } from "@aws-sdk/client-s3";
+import { exists, mkdir, writeFile } from "node:fs/promises";
+import { dirname, join } from "node:path";
+import {
+  DeleteObjectCommand,
+  GetObjectCommand,
+  paginateListObjectsV2,
+  S3,
+} from "@aws-sdk/client-s3";
 import { Upload } from "@aws-sdk/lib-storage";
 import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
 import { ConfiguredRetryStrategy } from "@smithy/util-retry";
+import { Glob } from "bun";
 import { lookup } from "mime-types";
-import { S3SyncClient } from "s3-sync-client";
+import { assert } from "shared/assert";
 import { env } from "../env";
-import type { PutObjectCommandInput } from "@aws-sdk/client-s3";
-import type { CommandInput } from "s3-sync-client";
 
-const retryStrategy = new ConfiguredRetryStrategy(5, 60_000);
+const retryStrategy = new ConfiguredRetryStrategy(
+  5,
+  (attempt) => attempt ** 2 * 1000,
+);
 
 const client = new S3({
   endpoint: env.S3_ENDPOINT,
@@ -18,88 +26,136 @@ const client = new S3({
     accessKeyId: env.S3_ACCESS_KEY,
     secretAccessKey: env.S3_SECRET_KEY,
   },
-  logger: console,
   retryStrategy,
 });
 
-const { sync } = new S3SyncClient({ client, retryStrategy });
-
-export async function syncFromS3(remotePath: string, localPath: string) {
-  await sync(`s3://${env.S3_BUCKET}/${remotePath}`, localPath);
-}
-
-export async function syncToS3(
+export async function s3UploadFile(
   localPath: string,
   remotePath: string,
-  options?: {
-    del?: boolean;
-    public?: boolean;
-    concurrency?: number;
-  },
+  aclPublic: boolean,
 ) {
-  const commandInput: CommandInput<PutObjectCommandInput> = (input) => {
-    let contentType: string | undefined;
-    if (input.Key) {
-      contentType = lookup(input.Key) || "binary/octet-stream";
-    }
-    return {
-      ContentType: contentType,
-      ACL: options?.public ? "public-read" : "private",
-    };
-  };
-
-  await sync(localPath, `s3://${env.S3_BUCKET}/${remotePath}`, {
-    del: options?.del,
-    commandInput,
-    maxConcurrentTransfers: options?.concurrency,
+  const upload = new Upload({
+    client,
+    params: {
+      Body: Bun.file(localPath).stream(),
+      ContentType: lookup(localPath) || "binary/octet-stream",
+      Bucket: env.S3_BUCKET,
+      Key: remotePath,
+      ACL: aclPublic ? "public-read" : "private",
+    },
   });
+  await upload.done();
 }
 
-type UploadToS3File =
-  | { type: "json"; data: object }
-  | { type: "local"; path: string };
+async function s3DownloadFile(remotePath: string, localPath: string) {
+  const command = new GetObjectCommand({
+    Bucket: env.S3_BUCKET,
+    Key: remotePath,
+  });
 
-export async function uploadToS3(
-  remoteFilePath: string,
-  file: UploadToS3File,
-  onProgress?: (value: number) => void,
+  const { Body } = await client.send(command);
+  assert(Body);
+
+  await writeFile(localPath, Body.transformToWebStream());
+}
+
+export async function s3DownloadFolder(remotePath: string, localPath: string) {
+  const paginatedListObjects = paginateListObjectsV2(
+    { client },
+    {
+      Bucket: env.S3_BUCKET,
+      Prefix: remotePath,
+    },
+  );
+
+  const filePaths: string[] = [];
+
+  for await (const data of paginatedListObjects) {
+    data.Contents?.forEach((content) => {
+      if (content.Key) {
+        filePaths.push(content.Key);
+      }
+    });
+  }
+
+  for (const filePath of filePaths) {
+    const localFilePath = join(
+      localPath,
+      filePath.substring(remotePath.length + 1),
+    );
+
+    const localFilePathDir = dirname(localFilePath);
+    const folderExists = await exists(localFilePathDir);
+    if (!folderExists) {
+      await mkdir(localFilePathDir, { recursive: true });
+    }
+
+    await s3DownloadFile(filePath, localFilePath);
+  }
+}
+
+export async function s3UploadFolder(
+  localPath: string,
+  remotePath: string,
+  aclPublic: boolean,
 ) {
-  let params: Omit<PutObjectCommandInput, "Bucket" | "Key"> | undefined;
-
-  switch (file.type) {
-    case "json":
-      params = {
-        Body: JSON.stringify(file.data, null, 2),
-        ContentType: "application/json",
-      };
-      break;
-    case "local":
-      params = {
-        Body: createReadStream(file.path),
-      };
-      break;
-    default:
-      return;
+  await s3DeleteFolder(remotePath);
+
+  const glob = new Glob("**/*");
+
+  const files: string[] = [];
+  for await (const file of glob.scan(localPath)) {
+    files.push(file);
   }
 
+  for (const file of files) {
+    await s3UploadFile(
+      `${localPath}/${file}`,
+      `${remotePath}/${file}`,
+      aclPublic,
+    );
+  }
+}
+
+export async function s3UploadJson(data: object, remotePath: string) {
   const upload = new Upload({
     client,
     params: {
-      ...params,
+      Body: JSON.stringify(data, null, 2),
+      ContentType: "application/json",
       Bucket: env.S3_BUCKET,
-      Key: remoteFilePath,
+      Key: remotePath,
     },
   });
+  await upload.done();
+}
 
-  upload.on("httpUploadProgress", (event) => {
-    if (event.loaded === undefined || event.total === undefined) {
-      return;
-    }
-    const value = Math.round((event.loaded / event.total) * 100);
-    onProgress?.(value);
-  });
+async function s3DeleteFolder(remotePath: string) {
+  const paginatedListObjects = paginateListObjectsV2(
+    { client },
+    {
+      Bucket: env.S3_BUCKET,
+      Prefix: remotePath,
+    },
+  );
 
-  await upload.done();
+  const filePaths: string[] = [];
+
+  for await (const data of paginatedListObjects) {
+    data.Contents?.forEach((content) => {
+      if (content.Key) {
+        filePaths.push(content.Key);
+      }
+    });
+  }
+
+  for (const filePath of filePaths) {
+    const command = new DeleteObjectCommand({
+      Bucket: env.S3_BUCKET,
+      Key: filePath,
+    });
+    await client.send(command);
+  }
 }
 
 export async function getS3SignedUrl(
@@ -110,8 +166,6 @@ export async function getS3SignedUrl(
     Bucket: env.S3_BUCKET,
     Key: remoteFilePath,
   });
-  // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-  // @ts-ignore https://github.com/aws/aws-sdk-js-v3/issues/4451
   const url = await getSignedUrl(client, command, {
     expiresIn,
   });
diff --git a/apps/artisan/src/workers/ffmpeg.ts b/apps/artisan/src/workers/ffmpeg.ts
@@ -1,6 +1,6 @@
 import { ffmpeg } from "../lib/ffmpeg";
 import { mapInputToPublicUrl } from "../lib/file-helpers";
-import { uploadToS3 } from "../lib/s3";
+import { s3UploadFile } from "../lib/s3";
 import type { FfmpegData, FfmpegResult, Stream, WorkerCallback } from "bolt";
 
 export const ffmpegCallback: WorkerCallback<FfmpegData, FfmpegResult> = async ({
@@ -55,15 +55,10 @@ export const ffmpegCallback: WorkerCallback<FfmpegData, FfmpegResult> = async ({
     `Uploading ${outDir}/${name} to transcode/${job.data.assetId}/${name}`,
   );
 
-  await uploadToS3(
+  await s3UploadFile(
+    `${outDir}/${name}`,
     `transcode/${job.data.assetId}/${name}`,
-    {
-      type: "local",
-      path: `${outDir}/${name}`,
-    },
-    (value) => {
-      progressTracker.set("upload", value);
-    },
+    false,
   );
 
   return {
diff --git a/apps/artisan/src/workers/package.ts b/apps/artisan/src/workers/package.ts
@@ -2,7 +2,7 @@ import { addToQueue, outcomeQueue } from "bolt";
 import { execa } from "execa";
 import parseFilePath from "parse-filepath";
 import { getBinaryPath, getMetaStruct } from "../lib/file-helpers";
-import { syncFromS3, syncToS3 } from "../lib/s3";
+import { s3DownloadFolder, s3UploadFolder } from "../lib/s3";
 import type {
   PackageData,
   PackageResult,
@@ -57,7 +57,7 @@ export const packageCallback: WorkerCallback<
 async function handleStepInitial(job: Job<PackageData>, dir: WorkerDir) {
   const inDir = await dir.createTempDir();
 
-  await syncFromS3(`transcode/${job.data.assetId}`, inDir);
+  await s3DownloadFolder(`transcode/${job.data.assetId}`, inDir);
 
   job.log(`Synced folder in ${inDir}`);
 
@@ -146,10 +146,7 @@ async function handleStepInitial(job: Job<PackageData>, dir: WorkerDir) {
   const s3Dir = `package/${job.data.assetId}/${job.data.name}`;
   job.log(`Uploading to ${s3Dir}`);
 
-  await syncToS3(outDir, s3Dir, {
-    del: true,
-    concurrency: job.data.concurrency,
-  });
+  await s3UploadFolder(outDir, s3Dir, job.data.public);
 }
 
 async function handleJobOutcome(job: Job<PackageData>) {
diff --git a/apps/artisan/src/workers/transcode.ts b/apps/artisan/src/workers/transcode.ts
@@ -12,7 +12,7 @@ import {
   getDefaultAudioBitrate,
   getDefaultVideoBitrate,
 } from "../lib/default-values";
-import { uploadToS3 } from "../lib/s3";
+import { s3UploadJson } from "../lib/s3";
 import type { MetaStruct } from "../lib/file-helpers";
 import type {
   FfmpegResult,
@@ -162,10 +162,7 @@ async function handleStepMeta(job: Job<TranscodeData>, token?: string) {
 
   await job.log(`Writing meta.json (${JSON.stringify(meta)})`);
 
-  await uploadToS3(`transcode/${job.data.assetId}/meta.json`, {
-    type: "json",
-    data: meta,
-  });
+  await s3UploadJson(meta, `transcode/${job.data.assetId}/meta.json`);
 }
 
 async function handleStepOutcome(job: Job<TranscodeData>) {
diff --git a/bun.lockb b/bun.lockb
