wow

apps/backend/src/lib/s3.ts

@@ -0,0 +1,124 @@
+import { S3Client } from "bun";
+import {
+  HeadObjectCommand,
+  PutObjectCommand,
+  S3Client as AwsS3Client,
+} from "@aws-sdk/client-s3";
+import config from "@/../config.toml";
+
+const client = new S3Client({
+  accessKeyId: config.s3.access_key,
+  secretAccessKey: config.s3.secret_key,
+  bucket: config.s3.bucket,
+  endpoint: config.s3.endpoint,
+});
+
+const awsClient = new AwsS3Client({
+  region: "auto",
+  endpoint: config.s3.endpoint,
+  forcePathStyle: true,
+  credentials: {
+    accessKeyId: config.s3.access_key,
+    secretAccessKey: config.s3.secret_key,
+  },
+});
+
+function makeS3FileName(authorId: string, tweetId: string, mediaUrl: string, index: number) {
+  const rawName = mediaUrl.split("/").pop() || `media_${Date.now()}_${index}`;
+  const withoutQuery = rawName.split("?")[0]?.split("#")[0] || `media_${Date.now()}_${index}`;
+  const safeName = withoutQuery.replace(/[^a-zA-Z0-9._-]/g, "_");
+  return `twitter/${authorId}/${tweetId}/${safeName || `media_${Date.now()}_${index}`}`;
+}
+
+async function uploadToS3(fileName: string, mediaUrl: string, maxRetry = 3) {
+  async function existsInS3(key: string) {
+    try {
+      return await client.exists(key);
+    } catch {
+      try {
+        await awsClient.send(new HeadObjectCommand({
+          Bucket: config.s3.bucket,
+          Key: key,
+        }));
+        return true;
+      } catch {
+        return false;
+      }
+    }
+  }
+
+  async function writeToS3(key: string, body: Uint8Array, mediaType?: string | null) {
+    try {
+      await client.write(key, body);
+      return;
+    } catch (bunWriteError) {
+      console.warn(`[S3 bun write failed, fallback to aws-sdk] key=${key}`, bunWriteError);
+      await awsClient.send(new PutObjectCommand({
+        Bucket: config.s3.bucket,
+        Key: key,
+        Body: body,
+        ContentType: mediaType ?? undefined,
+      }));
+    }
+  }
+
+  async function recoverByPollingExists(reason: string) {
+    for (let probe = 1; probe <= 4; probe++) {
+      await Bun.sleep(probe * 600);
+      try {
+        if (await existsInS3(fileName)) {
+          console.warn(`[S3 upload recovered-${reason}] key=${fileName} probe=${probe}`);
+          return true;
+        }
+      } catch (existsError) {
+        console.error(`[S3 exists probe failed] key=${fileName} probe=${probe}`, existsError);
+      }
+    }
+
+    return false;
+  }
+
+  let lastError: unknown;
+
+  for (let attempt = 1; attempt <= maxRetry; attempt++) {
+    try {
+      const response = await fetch(mediaUrl);
+      if (!response.ok) {
+        throw new Error(`Failed to fetch media from ${mediaUrl}: ${response.status} ${response.statusText}`);
+      }
+
+      const arrayBuffer = await response.arrayBuffer();
+      const buffer = Buffer.from(arrayBuffer);
+      await writeToS3(fileName, buffer, response.headers.get("content-type"));
+      return;
+    } catch (error) {
+      lastError = error;
+      console.error(`[S3 upload attempt ${attempt}/${maxRetry}] key=${fileName} url=${mediaUrl}`, error);
+
+      const errorCode =
+        typeof error === "object" && error !== null && "code" in error
+          ? String((error as { code?: unknown }).code)
+          : "";
+
+      // Some S3 providers return UnknownError even when the object is eventually persisted.
+      if (errorCode === "UnknownError") {
+        if (await recoverByPollingExists("unknown")) {
+          return;
+        }
+      }
+
+      if (attempt < maxRetry) {
+        await Bun.sleep(attempt * 800);
+      }
+    }
+  }
+
+  // Final guard: do one last exists check before surfacing failure.
+  if (await recoverByPollingExists("final")) {
+    return;
+  }
+
+  throw lastError;
+}
+
+export { makeS3FileName, uploadToS3, client as s3Client };