upload method

Future<void> upload(List<String> sources, String targetPath, { required bool recursive, required String onConflict, required bool preserveTimestamps, required List<String> include, required List<String> exclude, required String batchId, Map<String, dynamic>? initialBatchState, required Future<void> saveStateCallback(Map<String, dynamic>), dynamic onFileProgress(String filename, int current, int total, int bytesUploaded, int totalBytes,)?, int maxWorkers = kDefaultFileConcurrency, })

Batch upload with resume, conflict handling, and progress tracking.

Implementation

Future<void> upload(
  List<String> sources,
  String targetPath, {
  required bool recursive,
  required String onConflict,
  required bool preserveTimestamps,
  required List<String> include,
  required List<String> exclude,
  required String batchId,
  Map<String, dynamic>? initialBatchState,
  required Future<void> Function(Map<String, dynamic>) saveStateCallback,
  Function(String filename, int current, int total, int bytesUploaded,
          int totalBytes)?
      onFileProgress,
  int maxWorkers = kDefaultFileConcurrency,
}) async {
  api.log("Upload target path: $targetPath");

  Map<String, dynamic> batchState;
  List<dynamic> tasks;

  if (initialBatchState != null) {
    print("🔄 Resuming batch...");
    batchState = initialBatchState;
    tasks = batchState['tasks'] as List<dynamic>;
  } else {
    print("🔍 Building task list...");
    tasks = [];

    for (final sourceArg in sources) {
      if (sourceArg.contains('*') ||
          sourceArg.contains('?') ||
          sourceArg.contains('[')) {
        final glob = Glob(sourceArg.replaceAll('\\', '/'));
        await for (final entity in glob.list()) {
          await _processEntityForUpload(entity, sourceArg, targetPath,
              recursive, include, exclude, tasks, preserveTimestamps);
        }
      } else {
        final type = await FileSystemEntity.type(sourceArg);
        if (type == FileSystemEntityType.directory) {
          await _processEntityForUpload(
              Directory(sourceArg),
              sourceArg,
              targetPath,
              recursive,
              include,
              exclude,
              tasks,
              preserveTimestamps);
        } else if (type == FileSystemEntityType.file) {
          await _processEntityForUpload(
              File(sourceArg),
              sourceArg,
              targetPath,
              recursive,
              include,
              exclude,
              tasks,
              preserveTimestamps);
        } else {
          api.log("⚠️ Source not found: $sourceArg");
        }
      }
    }

    batchState = {
      'operationType': 'upload',
      'targetRemotePath': targetPath,
      'tasks': tasks,
    };
    await saveStateCallback(batchState);
    print("📝 Task list: ${tasks.length} files");
  }

  int successCount = 0;
  int skippedCount = 0;
  int errorCount = 0;
  int completedPreviously = 0;
  int processed = 0;
  final totalTasks = tasks.length;

  // Step 2: number of whole FILES uploaded at once. Capped at the count of
  // pending tasks and floored at 1 — a single file, or maxWorkers<=1, keeps
  // the sequential path (no shared budget, no pre-creation).
  final pending = [
    for (final t in tasks)
      if ((t as Map<String, dynamic>)['status'] != 'completed') t
  ];
  final effectiveWorkers =
      max(1, min(maxWorkers, pending.isEmpty ? 1 : pending.length));

  // saveStateCallback is async and shared (whole batchState). With files
  // completing out of order it must be serialized — a 1-permit semaphore
  // mutex prevents two concurrent file writers from interleaving.
  final saveMutex = ChunkSemaphore(1);
  Future<void> saveState() async {
    await saveMutex.acquire();
    try {
      await saveStateCallback(batchState);
    } finally {
      saveMutex.release();
    }
  }

  void tally(String token) {
    switch (token) {
      case 'completed':
        successCount++;
        break;
      case 'skipped':
        skippedCount++;
        break;
      case 'error':
        errorCount++;
        break;
      case 'already':
        completedPreviously++;
        break;
    }
    processed++;
    if (!api.debugMode) {
      final pct = totalTasks > 0
          ? (processed / totalTasks * 100).toStringAsFixed(1)
          : '0.0';
      stdout.write('\rUp: $processed/$totalTasks files ($pct%)  ');
    }
  }

  if (effectiveWorkers <= 1) {
    // Sequential path: behaves exactly like the pre-Step-2 loop.
    for (final t in tasks) {
      tally(await _uploadTask(
        t as Map<String, dynamic>,
        parentMap: null,
        onConflict: onConflict,
        preserveTimestamps: preserveTimestamps,
        saveState: saveState,
        maxConcurrentChunks: kDefaultUploadConcurrency,
        globalChunkSlots: null,
      ));
    }
  } else {
    // Pre-create unique parent folders ONCE, before fan-out, so the shared
    // createFolderRecursive + cache-invalidation side effects can't race
    // across concurrent files (constraint 3).
    final parentMap = <String, Map<String, dynamic>>{};
    for (final t in pending) {
      final rp = p.dirname(t['remotePath'].toString()).replaceAll('\\', '/');
      if (parentMap.containsKey(rp)) continue;
      try {
        parentMap[rp] = await drive.createFolderRecursive(rp);
      } catch (e) {
        // Leave it unmapped; the per-task path retries and marks
        // error_parent — same outcome as the sequential path.
        api.log('Pre-create parent failed for $rp: $e');
      }
    }

    // ONE shared budget across files × chunks (constraint 2). Per-file chunk
    // concurrency is lowered too so no single file monopolizes the budget.
    final perFileChunks =
        max(1, kGlobalMaxInflightChunks ~/ effectiveWorkers);
    final globalChunkSlots = ChunkSemaphore(kGlobalMaxInflightChunks);
    print(
        "  🧵 Uploading ${pending.length} file(s) with $effectiveWorkers worker(s)");

    await runWithConcurrency(tasks, effectiveWorkers, (t) async {
      tally(await _uploadTask(
        t as Map<String, dynamic>,
        parentMap: parentMap,
        onConflict: onConflict,
        preserveTimestamps: preserveTimestamps,
        saveState: saveState,
        maxConcurrentChunks: perFileChunks,
        globalChunkSlots: globalChunkSlots,
      ));
    });
  }

  if (!api.debugMode) stdout.write('\n');

  print('=' * 40);
  print('📊 Upload Summary:');
  if (completedPreviously > 0) print('  ✅ Previous: $completedPreviously');
  print('  ✅ Uploaded: $successCount');
  print('  ⏭️  Skipped: $skippedCount');
  print('  ❌ Errors: $errorCount');
  print('=' * 40);

  if (errorCount > 0) throw Exception("Upload finished with errors");
}