feat(download): serve HTTP Range on the proxy route

lib/storage.ts

@@ -72,6 +72,35 @@ export class ObjectNotFoundError extends Error {
   }
 }
 
+/** Adapters throw it without `size`; `Storage.download` fills it in from the location. */
+export class RangeNotSatisfiableError extends Error {
+  constructor(
+    objectName: string,
+    public readonly size?: number,
+  ) {
+    super(`Range not satisfiable for ${objectName}${size === undefined ? '' : ` (size ${size})`}`)
+    this.name = 'RangeNotSatisfiableError'
+  }
+}
+
+/** An omitted `end` means open-ended; the backend resolves it against the object length. */
+export interface RangeRequest {
+  start: number
+  end?: number
+}
+
+export interface ByteRange {
+  start: number
+  end: number
+}
+
+/** `range` is set only when a range was requested and the adapter honoured it. */
+export interface DownloadStream {
+  stream: Readable
+  size?: number
+  range?: ByteRange
+}
+
 export class AlreadyExistsError extends Error {
   constructor(objectName: string) {
     super(`Already exists: ${objectName}`)
@@ -148,11 +177,17 @@ export class Storage {
     if (actualPartCount < location.partCount) throw new ObjectNotFoundError(partsFolder)
   }
 
-  private async downloadFromCacheEntryLocation(location: StorageLocation) {
-    if (location.mergedAt) return this.adapter.createDownloadStream(`${location.folderName}/merged`)
+  private async downloadFromCacheEntryLocation(
+    location: StorageLocation,
+    range?: RangeRequest,
+  ): Promise<DownloadStream> {
+    if (location.mergedAt)
+      return this.adapter.createDownloadStream(`${location.folderName}/merged`, range)
 
     await this.ensurePartsExist(location)
-    return Readable.from(this.streamParts(location))
+    // No `range`: an unmerged entry is concatenated from its parts as it is read, so
+    // there is nothing to seek into. Served whole under a 200, which Range allows.
+    return { stream: Readable.from(this.streamParts(location)) }
   }
 
   private async pumpPartsToStreams(
@@ -180,7 +215,7 @@ export class Storage {
     if (location.partsDeletedAt) throw new Error('No parts to feed for location with deleted parts')
 
     for (let i = 0; i < location.partCount; i++) {
-      const partStream = await this.adapter.createDownloadStream(
+      const { stream: partStream } = await this.adapter.createDownloadStream(
         `${location.folderName}/parts/${i}`,
       )
 
@@ -536,12 +571,13 @@ export class Storage {
     }
   }
 
-  async download(cacheEntryId: string): Promise<Readable | undefined> {
+  /** Keyed by location, not entry: a re-save repoints the entry, but a location's bytes never change. */
+  async download(locationId: string, range?: RangeRequest): Promise<DownloadStream | undefined> {
     const protectedLocation = await this.db.transaction().execute(async (tx) => {
       let query = tx
         .selectFrom('storage_locations')
         .innerJoin('cache_entries', 'cache_entries.locationId', 'storage_locations.id')
-        .where('cache_entries.id', '=', cacheEntryId)
+        .where('storage_locations.id', '=', locationId)
         .selectAll('storage_locations')
       if (env.DB_DRIVER !== 'sqlite') query = query.forUpdate()
       const storageLocation = await query.executeTakeFirst()
@@ -563,8 +599,8 @@ export class Storage {
 
     try {
       if (storageLocation.mergedAt) {
-        const stream = await this.downloadFromCacheEntryLocation(storageLocation)
-        return this.protectDownloadStream(stream, readerLeaseId)
+        const download = await this.downloadFromCacheEntryLocation(storageLocation, range)
+        return { ...download, stream: this.protectDownloadStream(download.stream, readerLeaseId) }
       }
 
       await this.ensurePartsExist(storageLocation)
@@ -580,24 +616,26 @@ export class Storage {
           }),
       )
       if (!merge) {
-        const stream = await this.downloadFromCacheEntryLocation(storageLocation)
-        return this.protectDownloadStream(stream, readerLeaseId)
+        const download = await this.downloadFromCacheEntryLocation(storageLocation, range)
+        return { ...download, stream: this.protectDownloadStream(download.stream, readerLeaseId) }
       }
 
       this.pumpPartsToStreams(storageLocation, responseStream, mergerStream).catch((err) => {
         responseStream.destroy(err)
         mergerStream.destroy(err)
         if (err instanceof ObjectNotFoundError)
-          logger.warn(`Stale cache entry ${cacheEntryId}: ${err.message}`)
+          logger.warn(`Stale storage location ${locationId}: ${err.message}`)
       })
 
-      return this.protectDownloadStream(responseStream, readerLeaseId)
+      return { stream: this.protectDownloadStream(responseStream, readerLeaseId) }
     } catch (err) {
       await releaseReaderLease(this.db, readerLeaseId)
       if (err instanceof ObjectNotFoundError) {
-        logger.warn(`Stale cache entry ${cacheEntryId}: ${err.message}`)
+        logger.warn(`Stale storage location ${locationId}: ${err.message}`)
         return
       }
+      if (err instanceof RangeNotSatisfiableError)
+        throw new RangeNotSatisfiableError(locationId, storageLocation.sizeBytes ?? undefined)
       throw err
     }
   }
@@ -751,7 +789,7 @@ export class Storage {
         continue
       }
 
-      const defaultUrl = `${env.API_BASE_URL}/download/${cacheEntry.match.id}`
+      const defaultUrl = `${env.API_BASE_URL}/download/${cacheEntry.match.locationId}`
 
       if (!env.ENABLE_DIRECT_DOWNLOADS || !this.adapter.createDownloadUrl || !location.mergedAt)
         return {
@@ -801,7 +839,7 @@ export class Storage {
 export const getStorage = createSingletonPromise(async () => Storage.fromEnv())
 
 export interface StorageAdapter {
-  createDownloadStream(objectName: string): Promise<Readable>
+  createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream>
   /**
    * Uploads must be atomically visible: an object never exists partially, and
    * overwriting an object never disturbs active readers of the previous
@@ -1007,19 +1045,29 @@ class S3Adapter implements StorageAdapter {
     return deleted
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     try {
       const response = await this.s3.send(
         new GetObjectCommand({
           Bucket: this.bucket,
           Key: `${this.keyPrefix}/${objectName}`,
+          Range: range ? `bytes=${range.start}-${range.end ?? ''}` : undefined,
         }),
       )
       if (!response.Body) throw new Error('No body in S3 get object response')
 
-      return response.Body as Readable
+      const stream = response.Body as Readable
+      if (response.$metadata.httpStatusCode !== 206) return { stream, size: response.ContentLength }
+
+      const served = parseContentRange(response.ContentRange)
+      if (!served) {
+        stream.destroy()
+        throw new Error(`S3 answered 206 with unparseable Content-Range: ${response.ContentRange}`)
+      }
+      return { stream, size: served.size, range: { start: served.start, end: served.end } }
     } catch (err: any) {
       if (err.name === 'NoSuchKey') throw new ObjectNotFoundError(objectName)
+      if (err.name === 'InvalidRange') throw new RangeNotSatisfiableError(objectName)
       throw err
     }
   }
@@ -1172,14 +1220,20 @@ class FileSystemAdapter implements StorageAdapter {
     return folder
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     const filePath = this.safePath(objectName)
+    let size: number
     try {
-      await fs.access(filePath)
+      const stat = await fs.stat(filePath)
+      size = stat.size
     } catch {
       throw new ObjectNotFoundError(objectName)
     }
-    return createReadStream(filePath)
+    if (!range) return { stream: createReadStream(filePath), size }
+
+    const served = clampRange(range, size)
+    if (!served) throw new RangeNotSatisfiableError(objectName)
+    return { stream: createReadStream(filePath, served), size, range: served }
   }
 
   async objectExists(objectName: string) {
@@ -1329,11 +1383,22 @@ class GcsAdapter implements StorageAdapter {
     this.bucket = gcs.bucket(bucket)
   }
 
-  async createDownloadStream(objectName: string) {
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
     const file = this.bucket.file(`${this.keyPrefix}/${objectName}`)
-    const [exists] = await file.exists()
-    if (!exists) throw new ObjectNotFoundError(objectName)
-    return file.createReadStream()
+    // `getMetadata` proves existence and carries the size, so no second round-trip.
+    let size: number
+    try {
+      const [metadata] = await file.getMetadata()
+      size = Number(metadata.size)
+    } catch (err: any) {
+      if (err.code === 404) throw new ObjectNotFoundError(objectName)
+      throw err
+    }
+    if (!range) return { stream: file.createReadStream(), size }
+
+    const served = clampRange(range, size)
+    if (!served) throw new RangeNotSatisfiableError(objectName)
+    return { stream: file.createReadStream(served), size, range: served }
   }
 
   async objectExists(objectName: string) {
@@ -1534,10 +1599,35 @@ class AzBlobAdapter implements StorageAdapter {
     return deleted
   }
 
-  async createDownloadStream(objectName: string) {
-    const response = await this.blob(objectName).download()
+  async createDownloadStream(objectName: string, range?: RangeRequest): Promise<DownloadStream> {
+    let response
+    try {
+      response = await this.blob(objectName).download(
+        range?.start,
+        range?.end === undefined ? undefined : range.end - range.start + 1,
+      )
+    } catch (err: any) {
+      if (err.statusCode === 404) throw new ObjectNotFoundError(objectName)
+      if (err.statusCode === 416) throw new RangeNotSatisfiableError(objectName)
+      throw err
+    }
     if (!response.readableStreamBody) throw new Error(`No stream for blob "${objectName}"`)
-    return Readable.from(response.readableStreamBody)
+    const stream = Readable.from(response.readableStreamBody)
+
+    const served = parseContentRange(response.contentRange)
+    if (served) {
+      // Azurite answers a range past the end with an empty 206 where Azure answers 416.
+      if (served.start >= served.size) {
+        stream.destroy()
+        throw new RangeNotSatisfiableError(objectName)
+      }
+      return { stream, size: served.size, range: { start: served.start, end: served.end } }
+    }
+    // No Content-Range means the whole blob. The SDK drops the range header for `bytes=0-`,
+    // so that one request still got exactly what it asked for.
+    const size = response.contentLength
+    if (range?.start !== 0 || range.end !== undefined || !size) return { stream, size }
+    return { stream, size, range: { start: 0, end: size - 1 } }
   }
 
   async uploadStream(objectName: string, stream: Readable) {
@@ -1597,3 +1687,17 @@ class AzBlobAdapter implements StorageAdapter {
     return sign(objectName)
   }
 }
+
+function clampRange(range: RangeRequest, size: number): ByteRange | undefined {
+  if (range.start >= size) return
+  return { start: range.start, end: Math.min(range.end ?? size - 1, size - 1) }
+}
+
+const CONTENT_RANGE_RE = /^bytes (\d+)-(\d+)\/(\d+)$/
+
+function parseContentRange(header: string | undefined) {
+  if (!header) return
+  const m = CONTENT_RANGE_RE.exec(header)
+  if (!m) return
+  return { start: Number(m[1]), end: Number(m[2]), size: Number(m[3]) }
+}