Resumable file upload & download trong Flutter: hướng dẫn thực chiến về chunking

dependencies:
  dio: ^5.7.0
  path_provider: ^2.1.4
  crypto: ^3.0.5
  shared_preferences: ^2.3.2

class UploadSession {
  final String localPath;
  final String uploadId;
  final int totalBytes;
  final int chunkSize;
  int uploadedBytes;
  final String fileHash;

  UploadSession({
    required this.localPath,
    required this.uploadId,
    required this.totalBytes,
    required this.chunkSize,
    required this.uploadedBytes,
    required this.fileHash,
  });

  Map<String, dynamic> toJson() => {
        'localPath': localPath,
        'uploadId': uploadId,
        'totalBytes': totalBytes,
        'chunkSize': chunkSize,
        'uploadedBytes': uploadedBytes,
        'fileHash': fileHash,
      };

  factory UploadSession.fromJson(Map<String, dynamic> j) => UploadSession(
        localPath: j['localPath'],
        uploadId: j['uploadId'],
        totalBytes: j['totalBytes'],
        chunkSize: j['chunkSize'],
        uploadedBytes: j['uploadedBytes'],
        fileHash: j['fileHash'],
      );
}

Future<Uint8List> _readChunk(File file, int start, int length) async {
  final raf = await file.open(mode: FileMode.read);
  try {
    await raf.setPosition(start);
    return await raf.read(length);
  } finally {
    await raf.close();
  }
}

class ChunkedUploader {
  final Dio _dio;
  final String _baseUrl;
  static const int _chunkSize = 2 * 1024 * 1024; // 2 MB

  ChunkedUploader(this._dio, this._baseUrl);

  Future<String> upload(
    File file, {
    required void Function(double progress) onProgress,
    CancelToken? cancelToken,
  }) async {
    final session = await _resumeOrInitiate(file);

    while (session.uploadedBytes < session.totalBytes) {
      final start = session.uploadedBytes;
      final end = (start + _chunkSize > session.totalBytes)
          ? session.totalBytes
          : start + _chunkSize;
      final length = end - start;

      final chunk = await _readChunk(file, start, length);

      await _uploadChunkWithRetry(
        session: session,
        chunk: chunk,
        start: start,
        end: end - 1, // Content-Range is inclusive
        cancelToken: cancelToken,
      );

      session.uploadedBytes = end;
      await _persist(session);
      onProgress(session.uploadedBytes / session.totalBytes);
    }

    final fileUrl = await _complete(session);
    await _clear(session);
    return fileUrl;
  }
}

Future<void> _uploadChunkWithRetry({
  required UploadSession session,
  required Uint8List chunk,
  required int start,
  required int end,
  CancelToken? cancelToken,
}) async {
  const maxAttempts = 5;
  for (var attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      await _dio.put(
        '$_baseUrl/uploads/${session.uploadId}',
        data: Stream.fromIterable([chunk]),
        options: Options(
          headers: {
            'Content-Range': 'bytes $start-$end/${session.totalBytes}',
            'Content-Length': chunk.length,
            'Content-Type': 'application/octet-stream',
          },
        ),
        cancelToken: cancelToken,
      );
      return;
    } on DioException catch (e) {
      if (e.type == DioExceptionType.cancel) rethrow;
      if (attempt == maxAttempts) rethrow;
      // Exponential backoff with jitter: 1s, 2s, 4s, 8s …
      final delay = Duration(
        milliseconds: 1000 * (1 << (attempt - 1)) +
            Random().nextInt(500),
      );
      await Future.delayed(delay);
    }
  }
}

Future<UploadSession> _resumeOrInitiate(File file) async {
  final existing = await _loadSession(file.path);
  if (existing != null) {
    // Trust but verify — ask the server how far we actually got.
    final resp = await _dio.head(
      '$_baseUrl/uploads/${existing.uploadId}',
    );
    final serverOffset =
        int.tryParse(resp.headers.value('x-upload-offset') ?? '') ?? 0;
    existing.uploadedBytes = serverOffset;
    return existing;
  }

  final hash = await _sha256(file);
  final resp = await _dio.post(
    '$_baseUrl/uploads',
    data: {
      'fileName': p.basename(file.path),
      'totalBytes': await file.length(),
      'sha256': hash,
    },
  );
  return UploadSession(
    localPath: file.path,
    uploadId: resp.data['uploadId'],
    totalBytes: await file.length(),
    chunkSize: _chunkSize,
    uploadedBytes: 0,
    fileHash: hash,
  );
}

class ChunkedDownloader {
  final Dio _dio;
  static const int _chunkSize = 2 * 1024 * 1024;

  ChunkedDownloader(this._dio);

  Future<File> download(
    String url,
    String savePath, {
    required void Function(double) onProgress,
    CancelToken? cancelToken,
  }) async {
    final partialFile = File('$savePath.part');
    int downloaded =
        await partialFile.exists() ? await partialFile.length() : 0;

    final headResp = await _dio.head(url);
    final totalBytes = int.parse(
      headResp.headers.value(Headers.contentLengthHeader)!,
    );

    if (downloaded >= totalBytes) {
      await partialFile.rename(savePath);
      return File(savePath);
    }

    final sink = partialFile.openWrite(mode: FileMode.append);

    try {
      while (downloaded < totalBytes) {
        final end = (downloaded + _chunkSize > totalBytes)
            ? totalBytes - 1
            : downloaded + _chunkSize - 1;

        final chunk = await _fetchChunkWithRetry(
          url, downloaded, end, cancelToken,
        );
        sink.add(chunk);
        await sink.flush();

        downloaded = end + 1;
        onProgress(downloaded / totalBytes);
      }
    } finally {
      await sink.close();
    }

    await partialFile.rename(savePath);
    return File(savePath);
  }

  Future<List<int>> _fetchChunkWithRetry(
    String url, int start, int end, CancelToken? cancelToken,
  ) async {
    const maxAttempts = 5;
    for (var attempt = 1; attempt <= maxAttempts; attempt++) {
      try {
        final resp = await _dio.get<List<int>>(
          url,
          options: Options(
            responseType: ResponseType.bytes,
            headers: {'Range': 'bytes=$start-$end'},
          ),
          cancelToken: cancelToken,
        );
        return resp.data!;
      } on DioException catch (e) {
        if (e.type == DioExceptionType.cancel) rethrow;
        if (attempt == maxAttempts) rethrow;
        await Future.delayed(
          Duration(milliseconds: 1000 * (1 << (attempt - 1))),
        );
      }
    }
    throw StateError('unreachable');
  }
}

Future<bool> _verify(File file, String expectedHash) async {
  final stream = file.openRead();
  final digest = await sha256.bind(stream).first;
  return digest.toString() == expectedHash;
}

final uploader = ChunkedUploader(dio, 'https://api.example.com');
final cancelToken = CancelToken();

try {
  final url = await uploader.upload(
    File('/path/to/big.zip'),
    onProgress: (p) => setState(() => _progress = p),
    cancelToken: cancelToken,
  );
  showSnack('Uploaded: $url');
} on DioException catch (e) {
  if (e.type == DioExceptionType.cancel) return;
  showSnack('Upload failed — tap to resume');
}