@s3/store: allow providing custom cache implementation (#552)

* feat: allow providing custom cache implementation

* fix: reuse configstores as the s3 cache

* rename: configstores to kvstores

Author: fenos

packages/s3-store/README.md

@@ -82,6 +82,11 @@ Once the expiration period has passed, the upload url will return a 410 Gone sta
 Some S3 providers don't support tagging objects.
 If you are using certain features like the expiration extension and your provider doesn't support tagging, you can set this option to `false` to disable tagging.
 
+#### `options.cache`
+
+An optional cache implementation. If not provided, the store will use an in-memory cache (`MemoryConfigStore`).
+When running multiple instances of the server, you need to provide a cache implementation that is shared between all instances like the `RedisConfigStore`.
+
 ## Extensions
 
 The tus protocol supports optional [extensions][]. Below is a table of the supported extensions in `@tus/s3-store`.

packages/s3-store/index.ts

@@ -7,22 +7,25 @@ import http from 'node:http'
 import AWS, {NoSuchKey, NotFound, S3, S3ClientConfig} from '@aws-sdk/client-s3'
 import debug from 'debug'
 
-import {DataStore, StreamSplitter, Upload} from '@tus/server'
-import {ERRORS, TUS_RESUMABLE} from '@tus/server'
+import {
+  DataStore,
+  StreamSplitter,
+  Upload,
+  ERRORS,
+  TUS_RESUMABLE,
+  KvStore,
+  MemoryKvStore,
+} from '@tus/server'
 
 const log = debug('tus-node-server:stores:s3store')
 
-function calcOffsetFromParts(parts?: Array<AWS.Part>) {
-  // @ts-expect-error not undefined
-  return parts && parts.length > 0 ? parts.reduce((a, b) => a + b.Size, 0) : 0
-}
-
 type Options = {
   // The preferred part size for parts send to S3. Can not be lower than 5MiB or more than 5GiB.
   // The server calculates the optimal part size, which takes this size into account,
   // but may increase it to not exceed the S3 10K parts limit.
   partSize?: number
   useTags?: boolean
+  cache?: KvStore<MetadataValue>
   expirationPeriodInMilliseconds?: number
   // Options to pass to the AWS S3 SDK.
   s3ClientConfig: S3ClientConfig & {bucket: string}
@@ -33,6 +36,12 @@ type MetadataValue = {
   'upload-id': string
   'tus-version': string
 }
+
+function calcOffsetFromParts(parts?: Array<AWS.Part>) {
+  // @ts-expect-error not undefined
+  return parts && parts.length > 0 ? parts.reduce((a, b) => a + b.Size, 0) : 0
+}
+
 // Implementation (based on https://github.com/tus/tusd/blob/master/s3store/s3store.go)
 //
 // Once a new tus upload is initiated, multiple objects in S3 are created:
@@ -68,7 +77,7 @@ type MetadataValue = {
 // to S3.
 export class S3Store extends DataStore {
   private bucket: string
-  private cache: Map<string, MetadataValue> = new Map()
+  private cache: KvStore<MetadataValue>
   private client: S3
   private preferredPartSize: number
   private expirationPeriodInMilliseconds = 0
@@ -93,6 +102,7 @@ export class S3Store extends DataStore {
     this.expirationPeriodInMilliseconds = options.expirationPeriodInMilliseconds ?? 0
     this.useTags = options.useTags ?? true
     this.client = new S3(restS3ClientConfig)
+    this.cache = options.cache ?? new MemoryKvStore<MetadataValue>()
   }
 
   protected shouldUseExpirationTags() {
@@ -152,8 +162,8 @@ export class S3Store extends DataStore {
    * HTTP calls to S3.
    */
   private async getMetadata(id: string): Promise<MetadataValue> {
-    const cached = this.cache.get(id)
-    if (cached?.file) {
+    const cached = await this.cache.get(id)
+    if (cached) {
       return cached
     }
 
@@ -162,7 +172,7 @@ export class S3Store extends DataStore {
       Key: this.infoKey(id),
     })
     const file = JSON.parse((await Body?.transformToString()) as string)
-    this.cache.set(id, {
+    const metadata: MetadataValue = {
       'tus-version': Metadata?.['tus-version'] as string,
       'upload-id': Metadata?.['upload-id'] as string,
       file: new Upload({
@@ -172,8 +182,9 @@ export class S3Store extends DataStore {
         metadata: file.metadata,
         creation_date: file.creation_date,
       }),
-    })
-    return this.cache.get(id) as MetadataValue
+    }
+    await this.cache.set(id, metadata)
+    return metadata
   }
 
   private infoKey(id: string) {
@@ -423,10 +434,12 @@ export class S3Store extends DataStore {
     id: string,
     partNumberMarker?: string
   ): Promise<Array<AWS.Part>> {
+    const metadata = await this.getMetadata(id)
+
     const params: AWS.ListPartsCommandInput = {
       Bucket: this.bucket,
       Key: id,
-      UploadId: this.cache.get(id)?.['upload-id'],
+      UploadId: metadata['upload-id'],
       PartNumberMarker: partNumberMarker,
     }
 
@@ -450,9 +463,9 @@ export class S3Store extends DataStore {
   /**
    * Removes cached data for a given file.
    */
-  private clearCache(id: string) {
+  private async clearCache(id: string) {
     log(`[${id}] removing cached data`)
-    this.cache.delete(id)
+    await this.cache.delete(id)
   }
 
   private calcOptimalPartSize(size?: number): number {
@@ -546,7 +559,7 @@ export class S3Store extends DataStore {
         const parts = await this.retrieveParts(id)
         await this.finishMultipartUpload(metadata, parts)
         await this.completeMetadata(metadata.file)
-        this.clearCache(id)
+        await this.clearCache(id)
       } catch (error) {
         log(`[${id}] failed to finish upload`, error)
         throw error
@@ -579,8 +592,7 @@ export class S3Store extends DataStore {
       // Spaces, can also return NoSuchKey.
       if (error.Code === 'NoSuchUpload' || error.Code === 'NoSuchKey') {
         return new Upload({
-          id,
-          ...this.cache.get(id)?.file,
+          ...metadata.file,
           offset: metadata.file.size as number,
           size: metadata.file.size,
           metadata: metadata.file.metadata,
@@ -594,8 +606,7 @@ export class S3Store extends DataStore {
     const incompletePartSize = await this.getIncompletePartSize(id)
 
     return new Upload({
-      id,
-      ...this.cache.get(id)?.file,
+      ...metadata.file,
       offset: offset + (incompletePartSize ?? 0),
       size: metadata.file.size,
     })

packages/server/package.json

@@ -40,6 +40,9 @@
     "tsconfig": "*",
     "typescript": "^5.3.3"
   },
+  "optionalDependencies": {
+    "@redis/client": "^1.5.13"
+  },
   "engines": {
     "node": ">=16"
   }

packages/server/src/index.ts

@@ -3,3 +3,4 @@ export * from './types'
 export * from './models'
 export * from './lockers'
 export * from './constants'
+export * from './kvstores'

packages/server/src/kvstores/FileKvStore.ts

@@ -0,0 +1,48 @@
+import fs from 'node:fs/promises'
+import path from 'node:path'
+
+import {KvStore} from './Types'
+import {Upload} from '../models'
+
+/**
+ * FileConfigstore writes the `Upload` JSON metadata to disk next the uploaded file itself.
+ * It uses a queue which only processes one operation at a time to prevent unsafe concurrent access.
+ */
+export class FileKvStore<T = Upload> implements KvStore<T> {
+  directory: string
+
+  constructor(path: string) {
+    this.directory = path
+  }
+
+  async get(key: string): Promise<T | undefined> {
+    try {
+      const buffer = await fs.readFile(this.resolve(key), 'utf8')
+      return JSON.parse(buffer as string)
+    } catch {
+      return undefined
+    }
+  }
+
+  async set(key: string, value: T): Promise<void> {
+    await fs.writeFile(this.resolve(key), JSON.stringify(value))
+  }
+
+  async delete(key: string): Promise<void> {
+    await fs.rm(this.resolve(key))
+  }
+
+  async list(): Promise<Array<string>> {
+    const files = await fs.readdir(this.directory)
+    const sorted = files.sort((a, b) => a.localeCompare(b))
+    const name = (file: string) => path.basename(file, '.json')
+    // To only return tus file IDs we check if the file has a corresponding JSON info file
+    return sorted.filter(
+      (file, idx) => idx < sorted.length - 1 && name(file) === name(sorted[idx + 1])
+    )
+  }
+
+  private resolve(key: string): string {
+    return path.resolve(this.directory, `${key}.json`)
+  }
+}

packages/server/src/kvstores/MemoryKvStore.ts

@@ -0,0 +1,26 @@
+import {Upload} from '../models'
+import {KvStore} from './Types'
+
+/**
+ * Memory based configstore.
+ * Used mostly for unit tests.
+ */
+export class MemoryKvStore<T = Upload> implements KvStore<T> {
+  data: Map<string, T> = new Map()
+
+  async get(key: string): Promise<T | undefined> {
+    return this.data.get(key)
+  }
+
+  async set(key: string, value: T): Promise<void> {
+    this.data.set(key, value)
+  }
+
+  async delete(key: string): Promise<void> {
+    this.data.delete(key)
+  }
+
+  async list(): Promise<Array<string>> {
+    return [...this.data.keys()]
+  }
+}

packages/server/src/kvstores/RedisKvStore.ts

@@ -0,0 +1,39 @@
+import {RedisClientType} from '@redis/client'
+import {KvStore} from './Types'
+import {Upload} from '../models'
+
+/**
+ * Redis based configstore.
+ *
+ * @author Mitja Puzigaća <[email protected]>
+ */
+export class RedisKvStore<T = Upload> implements KvStore<T> {
+  constructor(private redis: RedisClientType, private prefix: string = '') {
+    this.redis = redis
+    this.prefix = prefix
+  }
+
+  async get(key: string): Promise<T | undefined> {
+    return this.deserializeValue(await this.redis.get(this.prefix + key))
+  }
+
+  async set(key: string, value: T): Promise<void> {
+    await this.redis.set(this.prefix + key, this.serializeValue(value))
+  }
+
+  async delete(key: string): Promise<void> {
+    await this.redis.del(this.prefix + key)
+  }
+
+  async list(): Promise<Array<string>> {
+    return this.redis.keys(this.prefix + '*')
+  }
+
+  private serializeValue(value: T): string {
+    return JSON.stringify(value)
+  }
+
+  private deserializeValue(buffer: string | null): T | undefined {
+    return buffer ? JSON.parse(buffer) : undefined
+  }
+}

packages/server/src/kvstores/Types.ts

@@ -0,0 +1,9 @@
+import {Upload} from '../models'
+
+export interface KvStore<T = Upload> {
+  get(key: string): Promise<T | undefined>
+  set(key: string, value: T): Promise<void>
+  delete(key: string): Promise<void>
+
+  list?(): Promise<Array<string>>
+}

packages/server/src/kvstores/index.ts

@@ -0,0 +1,4 @@
+export {FileKvStore} from './FileKvStore'
+export {MemoryKvStore} from './MemoryKvStore'
+export {RedisKvStore} from './RedisKvStore'
+export {KvStore} from './Types'

yarn.lock

@@ -1735,6 +1735,7 @@ __metadata:
   version: 0.0.0-use.local
   resolution: "@tus/server@workspace:packages/server"
   dependencies:
+    "@redis/client": "npm:^1.5.13"
     "@types/debug": "npm:^4.1.12"
     "@types/mocha": "npm:^10.0.6"
     "@types/node": "npm:^20.11.5"
@@ -1751,6 +1752,9 @@ __metadata:
     ts-node: "npm:^10.9.2"
     tsconfig: "npm:*"
     typescript: "npm:^5.3.3"
+  dependenciesMeta:
+    "@redis/client":
+      optional: true
   languageName: unknown
   linkType: soft