feat: multipart blob sync support (#14138)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Flexible blob uploads: GRAPHQL, presigned, and multipart flows with
per‑part URLs, abort/complete operations, presigned proxy endpoints, and
nightly cleanup of expired pending uploads.

* **API / Schema**
* GraphQL additions: new types, mutations, enum and error to manage
upload lifecycle (create, complete, abort, get part URL).

* **Database**
* New blob status enum and columns (status, upload_id); listing now
defaults to completed blobs.

* **Localization**
  * Added user-facing message: "Blob is invalid."

* **Tests**
* Expanded unit and end‑to‑end coverage for upload flows, proxy
behavior, multipart and provider integrations.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/server/src/base/error/def.ts

@@ -506,6 +506,10 @@ export const USER_FRIENDLY_ERRORS = {
     message: ({ spaceId, blobId }) =>
       `Blob ${blobId} not found in Space ${spaceId}.`,
   },
+  blob_invalid: {
+    type: 'invalid_input',
+    message: 'Blob is invalid.',
+  },
   expect_to_publish_doc: {
     type: 'invalid_input',
     message: 'Expected to publish a doc, not a Space.',

packages/backend/server/src/base/error/errors.gen.ts

@@ -437,6 +437,12 @@ export class BlobNotFound extends UserFriendlyError {
   }
 }
 
+export class BlobInvalid extends UserFriendlyError {
+  constructor(message?: string) {
+    super('invalid_input', 'blob_invalid', message);
+  }
+}
+
 export class ExpectToPublishDoc extends UserFriendlyError {
   constructor(message?: string) {
     super('invalid_input', 'expect_to_publish_doc', message);
@@ -1166,6 +1172,7 @@ export enum ErrorNames {
   INVALID_HISTORY_TIMESTAMP,
   DOC_HISTORY_NOT_FOUND,
   BLOB_NOT_FOUND,
+  BLOB_INVALID,
   EXPECT_TO_PUBLISH_DOC,
   EXPECT_TO_REVOKE_PUBLIC_DOC,
   EXPECT_TO_GRANT_DOC_USER_ROLES,

packages/backend/server/src/base/redis/instances.ts

@@ -19,8 +19,12 @@ class Redis extends IORedis implements OnModuleInit, OnModuleDestroy {
     this.on('error', this.errorHandler);
   }
 
-  onModuleDestroy() {
-    this.disconnect();
+  async onModuleDestroy() {
+    try {
+      await this.quit();
+    } catch {
+      this.disconnect();
+    }
   }
 
   override duplicate(override?: Partial<RedisOptions>): IORedis {

packages/backend/server/src/base/storage/__tests__/s3.spec.ts

@@ -0,0 +1,80 @@
+import test from 'ava';
+
+import { S3StorageProvider } from '../providers/s3';
+import { SIGNED_URL_EXPIRED } from '../providers/utils';
+
+const config = {
+  region: 'auto',
+  credentials: {
+    accessKeyId: 'test',
+    secretAccessKey: 'test',
+  },
+};
+
+function createProvider() {
+  return new S3StorageProvider(config, 'test-bucket');
+}
+
+test('presignPut should return url and headers', async t => {
+  const provider = createProvider();
+  const result = await provider.presignPut('key', {
+    contentType: 'text/plain',
+  });
+
+  t.truthy(result);
+  t.true(result!.url.length > 0);
+  t.true(result!.url.includes('X-Amz-Algorithm=AWS4-HMAC-SHA256'));
+  t.deepEqual(result!.headers, { 'Content-Type': 'text/plain' });
+  const now = Date.now();
+  t.true(result!.expiresAt.getTime() >= now + SIGNED_URL_EXPIRED * 1000 - 2000);
+  t.true(result!.expiresAt.getTime() <= now + SIGNED_URL_EXPIRED * 1000 + 2000);
+});
+
+test('presignUploadPart should return url', async t => {
+  const provider = createProvider();
+  const result = await provider.presignUploadPart('key', 'upload-1', 3);
+
+  t.truthy(result);
+  t.true(result!.url.length > 0);
+  t.true(result!.url.includes('X-Amz-Algorithm=AWS4-HMAC-SHA256'));
+});
+
+test('createMultipartUpload should return uploadId', async t => {
+  const provider = createProvider();
+  let receivedCommand: any;
+  const sendStub = async (command: any) => {
+    receivedCommand = command;
+    return { UploadId: 'upload-1' };
+  };
+  (provider as any).client = { send: sendStub };
+
+  const now = Date.now();
+  const result = await provider.createMultipartUpload('key', {
+    contentType: 'text/plain',
+  });
+
+  t.is(result?.uploadId, 'upload-1');
+  t.true(result!.expiresAt.getTime() >= now + SIGNED_URL_EXPIRED * 1000 - 2000);
+  t.true(result!.expiresAt.getTime() <= now + SIGNED_URL_EXPIRED * 1000 + 2000);
+  t.is(receivedCommand.input.Key, 'key');
+  t.is(receivedCommand.input.ContentType, 'text/plain');
+});
+
+test('completeMultipartUpload should order parts', async t => {
+  const provider = createProvider();
+  let called = false;
+  const sendStub = async (command: any) => {
+    called = true;
+    t.deepEqual(command.input.MultipartUpload.Parts, [
+      { ETag: 'a', PartNumber: 1 },
+      { ETag: 'b', PartNumber: 2 },
+    ]);
+  };
+  (provider as any).client = { send: sendStub };
+
+  await provider.completeMultipartUpload('key', 'upload-1', [
+    { partNumber: 2, etag: 'b' },
+    { partNumber: 1, etag: 'a' },
+  ]);
+  t.true(called);
+});

packages/backend/server/src/base/storage/providers/index.ts

@@ -118,7 +118,7 @@ export const StorageJSONSchema: JSONSchema = {
                 urlPrefix: {
                   type: 'string' as const,
                   description:
-                    'The presigned url prefix for the cloudflare r2 storage provider.\nsee https://developers.cloudflare.com/waf/custom-rules/use-cases/configure-token-authentication/ to configure it.\nExample value: "https://storage.example.com"\nExample rule: is_timed_hmac_valid_v0("your_secret", http.request.uri, 10800, http.request.timestamp.sec, 6)',
+                    'The custom domain URL prefix for the cloudflare r2 storage provider.\nWhen `enabled=true` and `urlPrefix` + `signKey` are provided, the server will:\n- Redirect GET requests to this custom domain with an HMAC token.\n- Return upload URLs under `/api/storage/*` for uploads.\nPresigned/upload proxy TTL is 1 hour.\nsee https://developers.cloudflare.com/waf/custom-rules/use-cases/configure-token-authentication/ to configure it.\nExample value: "https://storage.example.com"\nExample rule: is_timed_hmac_valid_v0("your_secret", http.request.uri, 10800, http.request.timestamp.sec, 6)',
                 },
                 signKey: {
                   type: 'string' as const,
@@ -135,4 +135,12 @@ export const StorageJSONSchema: JSONSchema = {
 };
 
 export type * from './provider';
-export { applyAttachHeaders, autoMetadata, sniffMime, toBuffer } from './utils';
+export {
+  applyAttachHeaders,
+  autoMetadata,
+  PROXY_MULTIPART_PATH,
+  PROXY_UPLOAD_PATH,
+  sniffMime,
+  STORAGE_PROXY_ROOT,
+  toBuffer,
+} from './utils';

packages/backend/server/src/base/storage/providers/provider.ts

@@ -25,12 +25,51 @@ export interface ListObjectsMetadata {
 export type BlobInputType = Buffer | Readable | string;
 export type BlobOutputType = Readable;
 
+export interface PresignedUpload {
+  url: string;
+  headers?: Record<string, string>;
+  expiresAt: Date;
+}
+
+export interface MultipartUploadInit {
+  uploadId: string;
+  expiresAt: Date;
+}
+
+export interface MultipartUploadPart {
+  partNumber: number;
+  etag: string;
+}
+
 export interface StorageProvider {
   put(
     key: string,
     body: BlobInputType,
     metadata?: PutObjectMetadata
   ): Promise<void>;
+  presignPut?(
+    key: string,
+    metadata?: PutObjectMetadata
+  ): Promise<PresignedUpload | undefined>;
+  createMultipartUpload?(
+    key: string,
+    metadata?: PutObjectMetadata
+  ): Promise<MultipartUploadInit | undefined>;
+  presignUploadPart?(
+    key: string,
+    uploadId: string,
+    partNumber: number
+  ): Promise<PresignedUpload | undefined>;
+  listMultipartUploadParts?(
+    key: string,
+    uploadId: string
+  ): Promise<MultipartUploadPart[] | undefined>;
+  completeMultipartUpload?(
+    key: string,
+    uploadId: string,
+    parts: MultipartUploadPart[]
+  ): Promise<void>;
+  abortMultipartUpload?(key: string, uploadId: string): Promise<void>;
   head(key: string): Promise<GetObjectMetadata | undefined>;
   get(
     key: string,

packages/backend/server/src/base/storage/providers/r2.ts

@@ -1,10 +1,20 @@
 import assert from 'node:assert';
 import { Readable } from 'node:stream';
 
+import { PutObjectCommand, UploadPartCommand } from '@aws-sdk/client-s3';
 import { Logger } from '@nestjs/common';
 
-import { GetObjectMetadata } from './provider';
+import {
+  GetObjectMetadata,
+  PresignedUpload,
+  PutObjectMetadata,
+} from './provider';
 import { S3StorageConfig, S3StorageProvider } from './s3';
+import {
+  PROXY_MULTIPART_PATH,
+  PROXY_UPLOAD_PATH,
+  SIGNED_URL_EXPIRED,
+} from './utils';
 
 export interface R2StorageConfig extends S3StorageConfig {
   accountId: string;
@@ -39,8 +49,24 @@ export class R2StorageProvider extends S3StorageProvider {
     this.key = this.encoder.encode(config.usePresignedURL?.signKey ?? '');
   }
 
-  private async signUrl(url: URL): Promise<string> {
-    const timestamp = Math.floor(Date.now() / 1000);
+  private get shouldUseProxyUpload() {
+    const { usePresignedURL } = this.config;
+    return (
+      !!usePresignedURL?.enabled &&
+      !!usePresignedURL.signKey &&
+      this.key.length > 0
+    );
+  }
+
+  private parseWorkspaceKey(fullKey: string) {
+    const [workspaceId, ...rest] = fullKey.split('/');
+    if (!workspaceId || rest.length !== 1) {
+      return null;
+    }
+    return { workspaceId, key: rest.join('/') };
+  }
+
+  private async signPayload(payload: string) {
     const key = await crypto.subtle.importKey(
       'raw',
       this.key,
@@ -51,14 +77,140 @@ export class R2StorageProvider extends S3StorageProvider {
     const mac = await crypto.subtle.sign(
       'HMAC',
       key,
-      this.encoder.encode(`${url.pathname}${timestamp}`)
+      this.encoder.encode(payload)
     );
 
-    const base64Mac = Buffer.from(mac).toString('base64');
+    return Buffer.from(mac).toString('base64');
+  }
+
+  private async signUrl(url: URL): Promise<string> {
+    const timestamp = Math.floor(Date.now() / 1000);
+    const base64Mac = await this.signPayload(`${url.pathname}${timestamp}`);
     url.searchParams.set('sign', `${timestamp}-${base64Mac}`);
     return url.toString();
   }
 
+  private async createProxyUrl(
+    path: string,
+    canonicalFields: (string | number | undefined)[],
+    query: Record<string, string | number | undefined>
+  ) {
+    const exp = Math.floor(Date.now() / 1000) + SIGNED_URL_EXPIRED;
+    const canonical = [
+      path,
+      ...canonicalFields.map(field =>
+        field === undefined ? '' : field.toString()
+      ),
+      exp.toString(),
+    ].join('\n');
+    const token = await this.signPayload(canonical);
+
+    const url = new URL(`http://localhost${path}`);
+    for (const [key, value] of Object.entries(query)) {
+      if (value === undefined) continue;
+      url.searchParams.set(key, value.toString());
+    }
+    url.searchParams.set('exp', exp.toString());
+    url.searchParams.set('token', `${exp}-${token}`);
+
+    return { url: url.pathname + url.search, expiresAt: new Date(exp * 1000) };
+  }
+
+  override async presignPut(
+    key: string,
+    metadata: PutObjectMetadata = {}
+  ): Promise<PresignedUpload | undefined> {
+    if (!this.shouldUseProxyUpload) {
+      return super.presignPut(key, metadata);
+    }
+
+    const parsed = this.parseWorkspaceKey(key);
+    if (!parsed) {
+      return super.presignPut(key, metadata);
+    }
+
+    const contentType = metadata.contentType ?? 'application/octet-stream';
+    const { url, expiresAt } = await this.createProxyUrl(
+      PROXY_UPLOAD_PATH,
+      [parsed.workspaceId, parsed.key, contentType, metadata.contentLength],
+      {
+        workspaceId: parsed.workspaceId,
+        key: parsed.key,
+        contentType,
+        contentLength: metadata.contentLength,
+      }
+    );
+
+    return {
+      url,
+      headers: { 'Content-Type': contentType },
+      expiresAt,
+    };
+  }
+
+  override async presignUploadPart(
+    key: string,
+    uploadId: string,
+    partNumber: number
+  ): Promise<PresignedUpload | undefined> {
+    if (!this.shouldUseProxyUpload) {
+      return super.presignUploadPart(key, uploadId, partNumber);
+    }
+
+    const parsed = this.parseWorkspaceKey(key);
+    if (!parsed) {
+      return super.presignUploadPart(key, uploadId, partNumber);
+    }
+
+    return this.createProxyUrl(
+      PROXY_MULTIPART_PATH,
+      [parsed.workspaceId, parsed.key, uploadId, partNumber],
+      {
+        workspaceId: parsed.workspaceId,
+        key: parsed.key,
+        uploadId,
+        partNumber,
+      }
+    );
+  }
+
+  async proxyPutObject(
+    key: string,
+    body: Readable | Buffer | Uint8Array | string,
+    options: { contentType?: string; contentLength?: number } = {}
+  ) {
+    return this.client.send(
+      new PutObjectCommand({
+        Bucket: this.bucket,
+        Key: key,
+        Body: body,
+        ContentType: options.contentType,
+        ContentLength: options.contentLength,
+      })
+    );
+  }
+
+  async proxyUploadPart(
+    key: string,
+    uploadId: string,
+    partNumber: number,
+    body: Readable | Buffer | Uint8Array | string,
+    options: { contentLength?: number } = {}
+  ) {
+    const result = await this.client.send(
+      new UploadPartCommand({
+        Bucket: this.bucket,
+        Key: key,
+        UploadId: uploadId,
+        PartNumber: partNumber,
+        Body: body,
+        ContentLength: options.contentLength,
+      })
+    );
+
+    return result.ETag;
+  }
+
   override async get(
     key: string,
     signedUrl?: boolean

packages/backend/server/src/base/storage/providers/s3.ts

@@ -2,15 +2,21 @@
 import { Readable } from 'node:stream';
 
 import {
+  AbortMultipartUploadCommand,
+  CompleteMultipartUploadCommand,
+  CreateMultipartUploadCommand,
   DeleteObjectCommand,
   GetObjectCommand,
   HeadObjectCommand,
   ListObjectsV2Command,
+  ListPartsCommand,
   NoSuchKey,
+  NoSuchUpload,
   NotFound,
   PutObjectCommand,
   S3Client,
   S3ClientConfig,
+  UploadPartCommand,
 } from '@aws-sdk/client-s3';
 import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
 import { Logger } from '@nestjs/common';
@@ -19,6 +25,9 @@ import {
   BlobInputType,
   GetObjectMetadata,
   ListObjectsMetadata,
+  MultipartUploadInit,
+  MultipartUploadPart,
+  PresignedUpload,
   PutObjectMetadata,
   StorageProvider,
 } from './provider';
@@ -89,6 +98,196 @@ export class S3StorageProvider implements StorageProvider {
     }
   }
 
+  async presignPut(
+    key: string,
+    metadata: PutObjectMetadata = {}
+  ): Promise<PresignedUpload | undefined> {
+    try {
+      const contentType = metadata.contentType ?? 'application/octet-stream';
+      const url = await getSignedUrl(
+        this.client,
+        new PutObjectCommand({
+          Bucket: this.bucket,
+          Key: key,
+          ContentType: contentType,
+        }),
+        { expiresIn: SIGNED_URL_EXPIRED }
+      );
+
+      return {
+        url,
+        headers: { 'Content-Type': contentType },
+        expiresAt: new Date(Date.now() + SIGNED_URL_EXPIRED * 1000),
+      };
+    } catch (e) {
+      this.logger.error(
+        `Failed to presign put object (${JSON.stringify({
+          key,
+          bucket: this.bucket,
+          metadata,
+        })}`
+      );
+      throw e;
+    }
+  }
+
+  async createMultipartUpload(
+    key: string,
+    metadata: PutObjectMetadata = {}
+  ): Promise<MultipartUploadInit | undefined> {
+    try {
+      const contentType = metadata.contentType ?? 'application/octet-stream';
+      const response = await this.client.send(
+        new CreateMultipartUploadCommand({
+          Bucket: this.bucket,
+          Key: key,
+          ContentType: contentType,
+        })
+      );
+
+      if (!response.UploadId) {
+        return;
+      }
+
+      return {
+        uploadId: response.UploadId,
+        expiresAt: new Date(Date.now() + SIGNED_URL_EXPIRED * 1000),
+      };
+    } catch (e) {
+      this.logger.error(
+        `Failed to create multipart upload (${JSON.stringify({
+          key,
+          bucket: this.bucket,
+          metadata,
+        })}`
+      );
+      throw e;
+    }
+  }
+
+  async presignUploadPart(
+    key: string,
+    uploadId: string,
+    partNumber: number
+  ): Promise<PresignedUpload | undefined> {
+    try {
+      const url = await getSignedUrl(
+        this.client,
+        new UploadPartCommand({
+          Bucket: this.bucket,
+          Key: key,
+          UploadId: uploadId,
+          PartNumber: partNumber,
+        }),
+        { expiresIn: SIGNED_URL_EXPIRED }
+      );
+
+      return {
+        url,
+        expiresAt: new Date(Date.now() + SIGNED_URL_EXPIRED * 1000),
+      };
+    } catch (e) {
+      this.logger.error(
+        `Failed to presign upload part (${JSON.stringify({ key, bucket: this.bucket, uploadId, partNumber })}`
+      );
+      throw e;
+    }
+  }
+
+  async listMultipartUploadParts(
+    key: string,
+    uploadId: string
+  ): Promise<MultipartUploadPart[] | undefined> {
+    const parts: MultipartUploadPart[] = [];
+    let partNumberMarker: string | undefined;
+
+    try {
+      // ListParts is paginated by part number marker
+      // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListParts.html
+      // R2 follows S3 semantics here.
+      while (true) {
+        const response = await this.client.send(
+          new ListPartsCommand({
+            Bucket: this.bucket,
+            Key: key,
+            UploadId: uploadId,
+            PartNumberMarker: partNumberMarker,
+          })
+        );
+
+        for (const part of response.Parts ?? []) {
+          if (!part.PartNumber || !part.ETag) {
+            continue;
+          }
+          parts.push({ partNumber: part.PartNumber, etag: part.ETag });
+        }
+
+        if (!response.IsTruncated) {
+          break;
+        }
+
+        if (response.NextPartNumberMarker === undefined) {
+          break;
+        }
+
+        partNumberMarker = response.NextPartNumberMarker;
+      }
+
+      return parts;
+    } catch (e) {
+      // the upload may have been aborted/expired by provider lifecycle rules
+      if (e instanceof NoSuchUpload || e instanceof NotFound) {
+        return undefined;
+      }
+      this.logger.error(`Failed to list multipart upload parts for \`${key}\``);
+      throw e;
+    }
+  }
+
+  async completeMultipartUpload(
+    key: string,
+    uploadId: string,
+    parts: MultipartUploadPart[]
+  ): Promise<void> {
+    try {
+      const orderedParts = [...parts].sort(
+        (left, right) => left.partNumber - right.partNumber
+      );
+
+      await this.client.send(
+        new CompleteMultipartUploadCommand({
+          Bucket: this.bucket,
+          Key: key,
+          UploadId: uploadId,
+          MultipartUpload: {
+            Parts: orderedParts.map(part => ({
+              ETag: part.etag,
+              PartNumber: part.partNumber,
+            })),
+          },
+        })
+      );
+    } catch (e) {
+      this.logger.error(`Failed to complete multipart upload for \`${key}\``);
+      throw e;
+    }
+  }
+
+  async abortMultipartUpload(key: string, uploadId: string): Promise<void> {
+    try {
+      await this.client.send(
+        new AbortMultipartUploadCommand({
+          Bucket: this.bucket,
+          Key: key,
+          UploadId: uploadId,
+        })
+      );
+    } catch (e) {
+      this.logger.error(`Failed to abort multipart upload for \`${key}\``);
+      throw e;
+    }
+  }
+
   async head(key: string) {
     try {
       const obj = await this.client.send(
@@ -164,7 +363,7 @@ export class S3StorageProvider implements StorageProvider {
         body: obj.Body,
         metadata: {
           // always set when putting object
-          contentType: obj.ContentType!,
+          contentType: obj.ContentType ?? 'application/octet-stream',
           contentLength: obj.ContentLength!,
           lastModified: obj.LastModified!,
           checksumCRC32: obj.ChecksumCRC32,

packages/backend/server/src/base/storage/providers/utils.ts

@@ -94,3 +94,7 @@ export function sniffMime(
 }
 
 export const SIGNED_URL_EXPIRED = 60 * 60; // 1 hour
+
+export const STORAGE_PROXY_ROOT = '/api/storage';
+export const PROXY_UPLOAD_PATH = `${STORAGE_PROXY_ROOT}/upload`;
+export const PROXY_MULTIPART_PATH = `${STORAGE_PROXY_ROOT}/multipart`;