Merge pull request #147 from desci-labs/s3-data-up

Upload improvements, large size uploads

.env.example

@@ -55,9 +55,13 @@ THEGRAPH_API_URL=http://host.docker.internal:8000/subgraphs/name/nodes
 
 # set to true if need to send email
 SHOULD_SEND_EMAIL=
+SENDGRID_API_KEY=
+
+# S3 Bucket
+AWS_S3_BUCKET_NAME=
+AWS_S3_BUCKET_REGION=
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
-SENDGRID_API_KEY=
 
 # for faucet
 HOT_WALLET_KEY=

.env.test

@@ -56,6 +56,11 @@ SHOULD_SEND_EMAIL=
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
 
+# S3 Bucket
+AWS_S3_BUCKET_ARN=
+AWS_S3_BUCKET_NAME=
+AWS_S3_BUCKET_REGION=
+
 # for faucet
 HOT_WALLET_KEY=
 # https://cso-classifier.internal

desci-server/kubernetes/deployment.yaml

@@ -56,6 +56,8 @@ spec:
           export SHOULD_SEND_EMAIL=true
           export AWS_ACCESS_KEY_ID={{ .Data.AWS_ACCESS_KEY_ID }}
           export AWS_SECRET_ACCESS_KEY={{ .Data.AWS_SECRET_ACCESS_KEY }}
+          export AWS_S3_BUCKET_NAME={{ .Data.AWS_S3_BUCKET_NAME }}
+          export AWS_S3_BUCKET_REGION={{ .Data.AWS_S3_BUCKET_REGION }}
           export THEGRAPH_API_URL={{ .Data.THEGRAPH_API_URL }}
           export HOT_WALLET_KEY={{ .Data.HOT_WALLET_KEY }}
           export CSO_CLASSIFIER_API={{ .Data.CSO_CLASSIFIER_API }}

desci-server/kubernetes/deployment_demo.yaml

@@ -56,6 +56,8 @@ spec:
           export SHOULD_SEND_EMAIL=true
           export AWS_ACCESS_KEY_ID={{ .Data.AWS_ACCESS_KEY_ID }}
           export AWS_SECRET_ACCESS_KEY={{ .Data.AWS_SECRET_ACCESS_KEY }}
+          export AWS_S3_BUCKET_NAME={{ .Data.AWS_S3_BUCKET_NAME }}
+          export AWS_S3_BUCKET_REGION={{ .Data.AWS_S3_BUCKET_REGION }}
           export THEGRAPH_API_URL={{ .Data.THEGRAPH_API_URL }}
           export HOT_WALLET_KEY={{ .Data.HOT_WALLET_KEY }}
           export CSO_CLASSIFIER_API={{ .Data.CSO_CLASSIFIER_API }}

desci-server/kubernetes/deployment_dev.yaml

@@ -56,6 +56,8 @@ spec:
           export SHOULD_SEND_EMAIL=true
           export AWS_ACCESS_KEY_ID={{ .Data.AWS_ACCESS_KEY_ID }}
           export AWS_SECRET_ACCESS_KEY={{ .Data.AWS_SECRET_ACCESS_KEY }}
+          export AWS_S3_BUCKET_NAME={{ .Data.AWS_S3_BUCKET_NAME }}
+          export AWS_S3_BUCKET_REGION={{ .Data.AWS_S3_BUCKET_REGION }}
           export THEGRAPH_API_URL={{ .Data.THEGRAPH_API_URL }}
           export HOT_WALLET_KEY={{ .Data.HOT_WALLET_KEY }}
           export CSO_CLASSIFIER_API={{ .Data.CSO_CLASSIFIER_API }}

desci-server/package.json

@@ -82,6 +82,7 @@
     "mkdirp": "^1.0.4",
     "morgan": "^1.10.0",
     "multer": "^1.4.5-lts.1",
+    "multer-s3": "^3.0.1",
     "multiformats": "^9.6.4",
     "multihashes": "^4.0.3",
     "nebulus": "^0.0.5",
@@ -124,6 +125,7 @@
     "@types/jsonwebtoken": "^8.5.4",
     "@types/mocha": "^10.0.1",
     "@types/morgan": "^1.9.3",
+    "@types/multer-s3": "^3.0.1",
     "@types/node": "^16.4.13",
     "@types/supertest": "^2.0.12",
     "@types/validator": "^13.6.3",

desci-server/src/controllers/data/update.ts

@@ -28,6 +28,7 @@ import {
   pinDirectory,
   RecursiveLsResult,
 } from 'services/ipfs';
+import { fetchFileStreamFromS3, isS3Configured } from 'services/s3';
 import {
   arrayXor,
   calculateTotalZipUncompressedSize,
@@ -126,7 +127,8 @@ export const update = async (req: Request, res: Response<UpdateResponse | ErrorR
     return res.status(400).json({ error: 'failed' });
   }
 
-  const files = req.files as Express.Multer.File[];
+  // const files = req.files as Express.Multer.File[];
+  const files = req.files as any[];
   if (!arrayXor([externalUrl, files.length, newFolderName?.length]))
     return res
       .status(400)
@@ -259,14 +261,20 @@ export const update = async (req: Request, res: Response<UpdateResponse | ErrorR
   }
 
   //Pin the new files
-  const structuredFilesForPinning: IpfsDirStructuredInput[] = files.map((f: any) => {
-    return { path: f.originalname, content: f.buffer };
-  });
+  const structuredFilesForPinning: IpfsDirStructuredInput[] = await Promise.all(
+    files.map(async (f: any) => {
+      if (isS3Configured) {
+        const fileStream = await fetchFileStreamFromS3(f.key);
+        return { path: f.originalname, content: fileStream };
+      }
+      return { path: f.originalname, content: f.buffer };
+    }),
+  );
 
   if (structuredFilesForPinning.length || externalUrlFiles?.length) {
     const filesToPin = structuredFilesForPinning.length ? structuredFilesForPinning : externalUrlFiles;
     if (filesToPin.length) uploaded = await pinDirectory(filesToPin);
-    if (!uploaded.length) res.status(400).json({ error: 'Failed uploading to ipfs' });
+    if (!uploaded.length) return res.status(400).json({ error: 'Failed uploading to ipfs' });
     logger.info('[UPDATE DATASET] Pinned files: ', uploaded.length);
   }
 
@@ -289,7 +297,7 @@ export const update = async (req: Request, res: Response<UpdateResponse | ErrorR
   //New folder creation, add to uploaded
   if (newFolderName) {
     const newFolder = await pinDirectory([{ path: newFolderName + '/.nodeKeep', content: Buffer.from('') }]);
-    if (!newFolder.length) res.status(400).json({ error: 'Failed creating new folder' });
+    if (!newFolder.length) return res.status(400).json({ error: 'Failed creating new folder' });
     uploaded = newFolder;
   }
 

desci-server/src/routes/v1/data.ts

@@ -1,14 +1,35 @@
 import { Router } from 'express';
 import multer = require('multer');
+import multerS3 from 'multer-s3';
+import { v4 } from 'uuid';
 
 import { pubTree, retrieveTree, deleteData, update, renameData } from 'controllers/data';
 import { diffData } from 'controllers/data/diff';
 import { moveData } from 'controllers/data/move';
 import { updateExternalCid } from 'controllers/data/updateExternalCid';
 import { ensureUser } from 'middleware/ensureUser';
+import { isS3Configured, s3Client } from 'services/s3';
 
 const router = Router();
-const upload = multer({ preservePath: true });
+
+const upload = isS3Configured
+  ? multer({
+      preservePath: true,
+      storage: multerS3({
+        s3: s3Client,
+        bucket: process.env.AWS_S3_BUCKET_NAME,
+        key: (req, file, cb) => {
+          const userId = (req as any).user.id;
+          const { uuid, contextPath } = (req as any).body;
+          if (!uuid || !contextPath || !userId) {
+            cb(new Error('Missing required params to form key'));
+          }
+          const key = `${userId}*${uuid}/${v4()}`; // adjust for dir uploads, doesn't start with '/'
+          cb(null, key);
+        },
+      }),
+    })
+  : multer({ preservePath: true });
 
 router.post('/update', [ensureUser, upload.array('files')], update);
 router.post('/updateExternalCid', [ensureUser], updateExternalCid);

desci-server/src/services/ipfs.ts

@@ -260,7 +260,7 @@ export const downloadSingleFile = async (url: string): Promise<PdfComponentSingl
 
 export interface IpfsDirStructuredInput {
   path: string;
-  content: Buffer | Readable;
+  content: Buffer | Readable | ReadableStream;
 }
 
 export interface IpfsPinnedResult {

desci-server/src/services/s3.ts

@@ -0,0 +1,34 @@
+import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3';
+
+import parentLogger from 'logger';
+
+const logger = parentLogger.child({
+  module: 'Services::S3',
+});
+
+export const isS3Configured = process.env.AWS_S3_BUCKET_NAME && process.env.AWS_S3_BUCKET_REGION;
+
+export const s3Client = isS3Configured
+  ? new S3Client({
+      region: process.env.AWS_S3_BUCKET_REGION,
+      credentials: {
+        accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+        secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
+      },
+    })
+  : null;
+
+export async function fetchFileStreamFromS3(key: string): Promise<ReadableStream | null> {
+  const params = {
+    Bucket: process.env.AWS_S3_BUCKET_NAME,
+    Key: key,
+  };
+
+  try {
+    const data = await s3Client.send(new GetObjectCommand(params));
+    return data.Body as ReadableStream;
+  } catch (error) {
+    logger.error('Error fetching from S3:', error);
+    return null;
+  }
+}