https://chairkick.com/assets/lib/chunk_uploader-88333cc6.js

1// Resumable multipart upload against the recording upload-session API:
2// each part is signed by Rails, PUT straight to storage, then confirmed.
3// Shared by the live recorder (which streams MediaRecorder chunks in) and the
4// video file uploader (which slices a File into parts).
5
6export const SERVER_PART_BYTES = 5 * 1024 * 1024
7// Each part needs two Rails round-trips (sign + complete) wrapped around the S3
8// PUT, so a single in-flight part leaves the upload bandwidth idle during those
9// round-trips. Uploading several parts at once keeps the connection saturated.
10// Live recording shares CPU/bandwidth with the active encode, so it stays
11// conservative; callers with nothing else competing can pass a higher value.
12const PART_UPLOAD_CONCURRENCY = 4
13const PART_UPLOAD_ATTEMPTS = 3
14
15export class ChunkUploader {
16  constructor({ uploadSession, csrfToken, uploadedParts = new Map(), concurrency = PART_UPLOAD_CONCURRENCY, onProgress, onStatus, errorNote = "Your recording is saved locally." }) {
17    this.uploadSession = uploadSession
18    this.csrfToken = csrfToken
19    this.errorNote = errorNote
20    this.uploadedParts = new Map(uploadedParts)
21    this.concurrency = concurrency
22    this.onProgress = onProgress
23    this.onStatus = onStatus
24    this.bufferedChunks = []
25    this.bufferedBytes = 0
26    this.totalRecordedBytes = 0
27    this.uploadedBytes = 0
28    this.nextChunkIndex = 0
29    this.uploadQueue = []
30    this.activeUploads = new Map()
31    this.activeRequests = new Set()
32    this.partProgress = new Map()
33    this.idleWaiters = []
34    this.lastError = null
35    this.cancelled = false
36  }
37
38  handleRecorderChunk(chunk, recordedBytes) {
39    if (this.cancelled) return
40
41    this.totalRecordedBytes = recordedBytes
42    this.bufferedChunks.push(chunk)
43    this.bufferedBytes += chunk.size
44    if (this.bufferedBytes >= SERVER_PART_BYTES) this.flush()
45  }
46
47  flush(force = false) {
48    if (this.cancelled || this.bufferedBytes === 0) return
49    if (!force && this.bufferedBytes < SERVER_PART_BYTES) return
50
51    const part = new Blob(this.bufferedChunks, { type: this.bufferedChunks[0]?.type || "application/octet-stream" })
52    this.bufferedChunks = []
53    this.bufferedBytes = 0
54    this.enqueuePart(part)
55  }
56
57  enqueuePart(part) {
58    if (this.cancelled || this.lastError) return
59
60    const chunkIndex = this.nextChunkIndex
61    this.nextChunkIndex += 1
62
63    if (this.alreadyUploaded(chunkIndex, part)) {
64      this.uploadedBytes += part.size
65      this.reportProgress()
66      return
67    }
68
69    this.uploadQueue.push({ chunkIndex, part })
70    this.startQueuedUploads()
71  }
72
73  alreadyUploaded(chunkIndex, part) {
74    return this.uploadedParts.get(chunkIndex) === part.size
75  }
76
77  startQueuedUploads() {
78    if (this.cancelled || this.lastError) return
79
80    while (this.activeUploads.size < this.concurrency && this.uploadQueue.length > 0) {
81      const upload = this.uploadQueue.shift()
82      const promise = this.uploadPartWithRetry(upload.chunkIndex, upload.part)
83        .then(() => {
84          this.partProgress.delete(upload.chunkIndex)
85          this.uploadedBytes += upload.part.size
86          this.reportProgress()
87        })
88        .catch((error) => this.handleUploadError(error))
89        .finally(() => {
90          this.activeUploads.delete(upload.chunkIndex)
91          this.partProgress.delete(upload.chunkIndex)
92          this.startQueuedUploads()
93          this.resolveIdleWaiters()
94        })
95
96      this.activeUploads.set(upload.chunkIndex, promise)
97    }
98
99    this.resolveIdleWaiters()
100  }
101
102  async uploadChunks(chunks) {
103    this.totalRecordedBytes = chunks.reduce((total, chunk) => total + chunk.size, 0)
104    for (const chunk of chunks) {
105      this.bufferedChunks.push(chunk)
106      this.bufferedBytes += chunk.size
107      if (this.bufferedBytes >= SERVER_PART_BYTES) {
108        this.flush(true)
109      }
110    }
111    this.flush(true)
112
113    return await this.waitForIdle()
114  }
115
116  async waitForIdle() {
117    if (this.idle()) return this.idleResult()
118
119    return await new Promise((resolve, reject) => {
120      this.idleWaiters.push({ resolve, reject })
121      this.resolveIdleWaiters()
122    })
123  }
124
125  async uploadPartWithRetry(chunkIndex, part) {
126    let attempt = 0
127
128    while (attempt < PART_UPLOAD_ATTEMPTS) {
129      if (this.cancelled) throw new Error("Upload cancelled.")
130
131      attempt += 1

132      try {
133        await this.uploadPart(chunkIndex, part)
134        return
135      } catch (error) {
136        if (attempt >= PART_UPLOAD_ATTEMPTS) throw error
137        this.onStatus?.(`Upload interrupted. Retrying chunk ${chunkIndex + 1} (${attempt + 1}/${PART_UPLOAD_ATTEMPTS}).`)
138        await new Promise((resolve) => setTimeout(resolve, attempt * 500))
139      }
140    }
141  }
142
143  async uploadPart(chunkIndex, part) {
144    const signedUpload = await this.signPart(chunkIndex, part)
145    if (this.cancelled) throw new Error("Upload cancelled.")
146
147    const etag = await this.putPart(chunkIndex, part, signedUpload)
148    if (this.cancelled) throw new Error("Upload cancelled.")
149
150    await this.completePart(chunkIndex, part, etag)
151  }
152
153  putPart(chunkIndex, part, signedUpload) {
154    return new Promise((resolve, reject) => {
155      const request = new XMLHttpRequest()
156      this.activeRequests.add(request)
157
158      request.open("PUT", signedUpload.upload_url)
159      Object.entries(signedUpload.headers || {}).forEach(([name, value]) => {
160        request.setRequestHeader(name, value)
161      })
162
163      request.upload.onprogress = (event) => {
164        if (!event.lengthComputable) return
165
166        this.partProgress.set(chunkIndex, Math.min(event.loaded, part.size))
167        this.reportProgress()
168      }
169
170      request.onload = () => {
171        this.clearActiveRequest(request)
172        if (request.status >= 200 && request.status < 300) {
173          resolve(request.getResponseHeader("ETag"))
174        } else {
175          reject(new Error(`Chunk ${chunkIndex + 1} could not be uploaded to storage (${request.status}).`))
176        }
177      }
178      request.onerror = () => {
179        this.clearActiveRequest(request)
180        reject(new Error(`Chunk ${chunkIndex + 1} could not reach storage. Check your connection and try again.`))
181      }
182      request.onabort = () => {
183        this.clearActiveRequest(request)
184        reject(new Error(`Chunk ${chunkIndex + 1} upload was cancelled.`))
185      }
186
187      request.send(part)
188    })
189  }
190
191  async signPart(chunkIndex, part) {
192    const response = await fetch(this.uploadSession.chunk_sign_url_template.replace(":chunk_index", chunkIndex), {
193      method: "POST",
194      headers: {
195        "Accept": "application/json",
196        "Content-Type": "application/json",
197        "X-CSRF-Token": this.csrfToken
198      },
199      body: JSON.stringify({
200        byte_size: part.size,
201        content_type: part.type || "application/octet-stream"
202      })
203    })
204
205    if (!response.ok) {
206      const body = await response.json().catch(() => ({}))
207      throw new Error(body.error || `Chunk ${chunkIndex} could not be prepared.`)
208    }
209
210    return response.json()
211  }
212
213  async completePart(chunkIndex, part, etag) {
214    const response = await fetch(this.uploadSession.chunk_complete_url_template.replace(":chunk_index", chunkIndex), {
215      method: "POST",
216      headers: {
217        "Accept": "application/json",
218        "Content-Type": "application/json",
219        "X-CSRF-Token": this.csrfToken
220      },
221      body: JSON.stringify({
222        byte_size: part.size,
223        etag
224      })
225    })
226
227    if (!response.ok) {
228      const body = await response.json().catch(() => ({}))
229      throw new Error(body.error || `Chunk ${chunkIndex} could not be confirmed.`)
230    }
231
232    return response.json()
233  }
234
235  cancel() {
236    this.cancelled = true
237    this.uploadQueue = []
238    this.abortActiveRequests()
239    this.partProgress.clear()
240    this.bufferedChunks = []
241    this.bufferedBytes = 0
242    this.resolveIdleWaiters()
243  }
244
245  clearActiveRequest(request) {
246    this.activeRequests.delete(request)
247  }
248
249  abortActiveRequests() {
250    this.activeRequests.forEach((request) => request.abort())
251    this.activeRequests.clear()
252  }
253
254  handleUploadError(error) {
255    if (this.cancelled || this.lastError) return
256
257    this.lastError = error
258    this.uploadQueue = []
259    this.bufferedChunks = []
260    this.bufferedBytes = 0
261    this.abortActiveRequests()
262    this.onStatus?.([error?.message || "Upload paused.", this.errorNote].filter(Boolean).join(" "))
263  }
264
265  reportProgress() {
266    const inFlightBytes = Array.from(this.partProgress.values()).reduce((total, bytes) => total + bytes, 0)
267    const queuedBytes = this.uploadQueue.reduce((total, upload) => total + upload.part.size, 0)
268    const uploaded = this.uploadedBytes + inFlightBytes
269    const total = Math.max(this.totalRecordedBytes, uploaded + queuedBytes + this.bufferedBytes)
270
271    this.onProgress?.(uploaded, total)
272  }
273
274  idle() {
275    return this.uploadQueue.length === 0 && this.activeUploads.size === 0
276  }
277
278  idleResult() {
279    if (this.lastError) throw this.lastError
280
281    return { chunkCount: this.nextChunkIndex, uploadedBytes: this.uploadedBytes }
282  }
283
284  resolveIdleWaiters() {
285    if (!this.idle()) return
286
287    const waiters = this.idleWaiters.splice(0)

288    waiters.forEach(({ resolve, reject }) => {
289      try {
290        resolve(this.idleResult())
291      } catch (error) {
292        reject(error)
293      }
294    })
295  }
296}