uploadFileChunked method

Future<Map<String, String>> uploadFileChunked(File file, String parent, { String? fileUuid, String? creationTime, String? modificationTime, String? resumeUploadKey, int resumeFromChunk = 0, Set<int>? completedChunks, String? fileKey, int maxConcurrentChunks = kDefaultUploadConcurrency, ChunkSemaphore? globalChunkSlots, dynamic onProgress(int current, int total, int bytesUploaded, int totalBytes,)?, dynamic onUploadStart(String uuid, String uploadKey, String fileKey)?, void onChunksCompleted(Set<int> completed)?, })

Implementation

Future<Map<String, String>> uploadFileChunked(
  File file,
  String parent, {
  String? fileUuid,
  String? creationTime,
  String? modificationTime,
  String? resumeUploadKey,
  int resumeFromChunk = 0,
  Set<int>? completedChunks,
  String? fileKey,
  int maxConcurrentChunks = kDefaultUploadConcurrency,
  ChunkSemaphore? globalChunkSlots,
  Function(int current, int total, int bytesUploaded, int totalBytes)?
      onProgress,
  Function(String uuid, String uploadKey, String fileKey)? onUploadStart,
  void Function(Set<int> completed)? onChunksCompleted,
}) async {
  final name = p.basename(file.path);
  final size = await file.length();
  final uuid = fileUuid ?? crypto.uuid();
  final mk = masterKeys.last;
  if (mk.isEmpty) throw Exception('No master keys available');

  // Reuse the caller-supplied key when resuming so chunks uploaded across
  // attempts share one key (otherwise already-uploaded chunks would be
  // undecryptable). Generate a fresh key for new uploads.
  final fileKeyStr = fileKey ?? crypto.randomString(32);
  final fileKeyBytes = Uint8List.fromList(utf8.encode(fileKeyStr));

  var lastMod = modificationTime;
  if (lastMod == null && creationTime == null) {
    try {
      final stat = await file.stat();
      lastMod = stat.modified.millisecondsSinceEpoch.toString();
    } catch (_) {}
  }

  // Handle empty files
  if (size == 0) {
    api.log('Uploading empty file via /v3/upload/empty');

    final metaJson = json.encode({
      'name': name,
      'size': size,
      'mime': 'application/octet-stream',
      'key': fileKeyStr,
      'hash': '',
      'lastModified': lastMod != null
          ? int.tryParse(lastMod) ?? DateTime.now().millisecondsSinceEpoch
          : DateTime.now().millisecondsSinceEpoch,
    });

    final nameEncrypted = await crypto.encryptMetadata002(name, fileKeyStr);
    final sizeEncrypted =
        await crypto.encryptMetadata002(size.toString(), fileKeyStr);
    final mimeEncrypted = await crypto.encryptMetadata002(
        'application/octet-stream', fileKeyStr);
    final metadataEncrypted = await crypto.encryptMetadata002(metaJson, mk);
    final nameHashed = await crypto.hashFileName(name, masterKeys, email);

    await api.post('/v3/upload/empty', {
      'uuid': uuid,
      'name': nameEncrypted,
      'nameHashed': nameHashed,
      'size': sizeEncrypted,
      'parent': parent,
      'mime': mimeEncrypted,
      'metadata': metadataEncrypted,
      'version': 2,
    });

    cache.invalidate(parent);
    if (onProgress != null) onProgress(1, 1, 0, 0);

    return {'uuid': uuid, 'hash': '', 'size': '0'};
  }

  // Regular chunked upload
  final uploadKey = resumeUploadKey ?? crypto.randomString(32);

  // Resume is a SET of completed indices, not a high-water mark: with
  // concurrent uploads chunks finish out of order. resumeFromChunk (legacy)
  // folds in as a contiguous range; completedChunks carries an exact set.
  final done = <int>{...?completedChunks};
  if (resumeFromChunk > 0) {
    for (var i = 0; i < resumeFromChunk; i++) {
      done.add(i);
    }
  }

  if (onUploadStart != null && done.isEmpty) {
    onUploadStart(uuid, uploadKey, fileKeyStr);
  }

  final rm = crypto.randomString(32);
  const chunkSz = 1048576;
  final totalChunks = (size / chunkSz).ceil();
  final ingest = 'https://ingest.filen.io';

  // Running SHA-512 over the *plaintext* chunks, in order — cannot be
  // parallelized. A sequential producer reads + hashes each chunk in order
  // (cheap) and hands the slow network POST to the bounded pool.
  final digestSink = DigestSink();
  final byteSink = crypto_pkg.sha512.startChunkedConversion(digestSink);

  final completed = <int>{...done};

  // POST one already-encrypted chunk; throws on a non-200 response.
  Future<void> postChunk(int idx, Uint8List enc) async {
    final hashHex =
        HEX.encode(crypto_pkg.sha512.convert(enc).bytes).toLowerCase();
    final url = Uri.parse(
        '$ingest/v3/upload?uuid=$uuid&index=$idx&parent=$parent&uploadKey=$uploadKey&hash=$hashHex');
    final r = await api.client.post(url, body: enc, headers: {
      'Authorization': 'Bearer ${api.apiKey}'
    }).timeout(Duration(seconds: 30));
    if (r.statusCode != 200) {
      throw Exception('Chunk upload failed: ${r.statusCode} - ${r.body}');
    }
  }

  void reportProgress() {
    if (onProgress != null) {
      final n = completed.length;
      onProgress(n, totalChunks, min(n * chunkSz, size), size);
    }
    onChunksCompleted?.call(Set.of(completed));
  }

  ChunkUploadException fail(int idx, Object e) {
    api.log('Chunk $idx failed: $e');
    return ChunkUploadException(
      'Chunk $idx upload failed',
      fileUuid: uuid,
      uploadKey: uploadKey,
      lastSuccessfulChunk: contiguousCompletedMax(completed),
      completedChunks: Set.of(completed),
      fileKey: fileKeyStr,
      originalError: e,
    );
  }

  // Tiny files (and concurrency disabled) keep the simple sequential path:
  // no semaphore, no MemoryGate — nothing is spun up.
  final useConcurrency =
      maxConcurrentChunks > 1 && totalChunks > kSequentialChunkThreshold;

  final raf = await file.open();
  try {
    if (useConcurrency) {
      // N chunk Futures in flight, bounded by BOTH a count semaphore and the
      // byte-budget MemoryGate (≈ N×(plaintext+encrypted) live at once).
      final sem = ChunkSemaphore(maxConcurrentChunks);
      final inflight = <Future<void>>[];
      final errors = <MapEntry<int, Object>>[];

      var idx = 0;
      var off = 0;
      while (off < size) {
        final len = min(chunkSz, size - off);
        final bytes = await raf.read(len);
        byteSink.add(bytes); // in-order plaintext hash (sequential producer)
        off += len;
        final myIdx = idx++;
        if (done.contains(myIdx)) continue;
        if (errors.isNotEmpty) break;

        await sem.acquire(); // bound concurrency by count
        if (errors.isNotEmpty) {
          sem.release();
          break;
        }
        final enc = await crypto.encryptData(bytes, fileKeyBytes);
        final budget = bytes.length + enc.length; // plaintext + encrypted
        await memoryGate.acquire(budget); // bound concurrency by bytes

        inflight.add(() async {
          try {
            // Shared batch budget (Step 2): bound the number of chunk POSTs
            // in flight across the WHOLE batch, not just this file. No-op when
            // uploading a single file (globalChunkSlots == null).
            if (globalChunkSlots != null) await globalChunkSlots.acquire();
            try {
              await postChunk(myIdx, enc);
            } finally {
              if (globalChunkSlots != null) globalChunkSlots.release();
            }
            completed.add(myIdx);
            reportProgress();
          } catch (e) {
            errors.add(MapEntry(myIdx, e));
          } finally {
            memoryGate.release(budget);
            sem.release();
          }
        }());
      }

      await Future.wait(inflight); // join all in-flight workers
      if (errors.isNotEmpty) {
        errors.sort((a, b) => a.key.compareTo(b.key));
        throw fail(errors.first.key, errors.first.value);
      }
    } else {
      var idx = 0;
      var off = 0;
      while (off < size) {
        final len = min(chunkSz, size - off);
        final bytes = await raf.read(len);
        byteSink.add(bytes); // in-order plaintext hash
        off += len;
        final myIdx = idx++;
        if (done.contains(myIdx)) continue;

        final enc = await crypto.encryptData(bytes, fileKeyBytes);
        // Shared batch budget (Step 2): one permit per chunk in flight across
        // the batch. No-op for a lone file (globalChunkSlots == null).
        if (globalChunkSlots != null) await globalChunkSlots.acquire();
        try {
          await postChunk(myIdx, enc);
        } catch (e) {
          throw fail(myIdx, e);
        } finally {
          if (globalChunkSlots != null) globalChunkSlots.release();
        }
        completed.add(myIdx);
        reportProgress();
      }
    }

    print('');

    byteSink.close();
    final totalHash = HEX.encode(digestSink.value?.bytes ?? []).toLowerCase();

    final metaJsonWithHash = json.encode({
      'name': name,
      'size': size,
      'mime': 'application/octet-stream',
      'key': fileKeyStr,
      'hash': totalHash,
      'lastModified': lastMod != null
          ? int.tryParse(lastMod) ?? DateTime.now().millisecondsSinceEpoch
          : DateTime.now().millisecondsSinceEpoch,
    });

    final nameEncrypted = await crypto.encryptMetadata002(name, fileKeyStr);
    final sizeEncrypted =
        await crypto.encryptMetadata002(size.toString(), fileKeyStr);
    final mimeEncrypted = await crypto.encryptMetadata002(
        'application/octet-stream', fileKeyStr);
    final metadataEncryptedWithHash =
        await crypto.encryptMetadata002(metaJsonWithHash, mk);
    final nameHashed = await crypto.hashFileName(name, masterKeys, email);

    await api.post('/v3/upload/done', {
      'uuid': uuid,
      'name': nameEncrypted,
      'nameHashed': nameHashed,
      'size': sizeEncrypted,
      'chunks': totalChunks,
      'mime': mimeEncrypted,
      'rm': rm,
      'metadata': metadataEncryptedWithHash,
      'version': 2,
      'uploadKey': uploadKey,
    });

    cache.invalidate(parent);
    return {'uuid': uuid, 'hash': totalHash, 'size': size.toString()};
  } finally {
    await raf.close();
  }
}