Design file upload

class UploadWorker(
    context: Context,
    params: WorkerParameters,
    private val api: UploadsApi,
    private val http: OkHttpClient,
    private val store: UploadStateStore,
) : CoroutineWorker(context, params) {

    override suspend fun doWork(): Result {
        val state = store.load(inputData.getString("uploadId")!!) ?: return Result.failure()
        setForeground(progressNotification(state))              // long-running: show progress

        return try {
            val done = api.completedParts(state.uploadId)        // what storage already has
            val missing = (1..state.partCount).filter { it !in done }
            for (grant in api.partUrls(state.uploadId, missing)) {
                val body = state.readPart(grant.partNumber).toRequestBody()
                val response = http.newCall(Request.Builder().url(grant.url).put(body).build()).execute()
                response.use { if (!it.isSuccessful) return Result.retry() }   // backoff, then resume
                store.markPartDone(state.uploadId, grant.partNumber)
            }
            api.complete(state.uploadId)
            Result.success()
        } catch (e: IOException) {
            Result.retry()
        }
    }
}

fun enqueueUpload(context: Context, uploadId: String) {
    val request = OneTimeWorkRequestBuilder<UploadWorker>()
        .setInputData(workDataOf("uploadId" to uploadId))
        .setConstraints(Constraints(requiredNetworkType = NetworkType.CONNECTED))
        .setBackoffCriteria(BackoffPolicy.EXPONENTIAL, 30, TimeUnit.SECONDS)
        .build()
    WorkManager.getInstance(context).enqueueUniqueWork("upload-$uploadId", ExistingWorkPolicy.KEEP, request)
}

// Server side: the endpoints the clients above call
@PostMapping("/uploads/{id}/parts")
public List<PartGrant> partUrls(@AuthenticationPrincipal UserId user, @PathVariable UploadId id,
                                @RequestBody List<Integer> partNumbers) {
    Upload upload = uploads.find(id, user).orElseThrow(NotFound::new);
    return partNumbers.stream().map(n -> new PartGrant(n, presigner.presignUploadPart(p -> p
            .signatureDuration(Duration.ofHours(1))
            .uploadPartRequest(r -> r.bucket(BUCKET).key(upload.key())
                .uploadId(upload.multipartId()).partNumber(n))).url().toString()))
        .toList();
}

@GetMapping("/uploads/{id}/parts")
public Set<Integer> completedParts(@AuthenticationPrincipal UserId user, @PathVariable UploadId id) {
    Upload upload = uploads.find(id, user).orElseThrow(NotFound::new);
    return s3.listPartsPaginator(r -> r.bucket(BUCKET).key(upload.key()).uploadId(upload.multipartId()))
        .parts().stream().map(Part::partNumber).collect(Collectors.toSet());
}

// Web client: upload missing parts, 3 at a time
async function resumeUpload(file: File, state: UploadState, api: UploadsApi) {
  const done = new Set(await api.completedParts(state.uploadId));
  const missing = range(1, state.partCount).filter((n) => !done.has(n));
  const grants = await api.partUrls(state.uploadId, missing);

  await runWithConcurrency(3, grants, async ({ partNumber, url }) => {
    const start = (partNumber - 1) * state.partSize;
    const blob = file.slice(start, start + state.partSize);
    const res = await fetch(url, { method: "PUT", body: blob });
    if (!res.ok) throw new Error(`Part ${partNumber} failed: ${res.status}`);
  });

  await api.complete(state.uploadId);   // server calls CompleteMultipartUpload, then verifies
}