fix streaming sync

Author: atticusofsparta

lib/main.dart

@@ -26,7 +26,6 @@ import 'package:ardrive/shared/blocs/banner/app_banner_bloc.dart';
 import 'package:ardrive/sharing/blocs/sharing_file_bloc.dart';
 import 'package:ardrive/sync/data/snapshot_validation_service.dart';
 import 'package:ardrive/sync/domain/repositories/sync_repository.dart';
-import 'package:ardrive/theme/theme_switcher_bloc.dart';
 import 'package:ardrive/theme/theme_switcher_state.dart';
 import 'package:ardrive/turbo/services/payment_service.dart';
 import 'package:ardrive/turbo/services/upload_service.dart';

lib/sync/domain/repositories/sync_repository.dart

@@ -38,7 +38,6 @@ import 'package:ardrive/utils/snapshots/snapshot_drive_history.dart';
 import 'package:ardrive/utils/snapshots/snapshot_item.dart';
 import 'package:ardrive_utils/ardrive_utils.dart';
 import 'package:ario_sdk/ario_sdk.dart';
-import 'package:pool/pool.dart';
 import 'package:arweave/arweave.dart';
 import 'package:cryptography/cryptography.dart';
 import 'package:drift/drift.dart';
@@ -608,6 +607,8 @@ class _SyncRepository implements SyncRepository {
       ownerAddress: ownerAddress,
     );
 
+    txFechedCallback?.call(drive.id, gqlDriveHistory.txCount);
+
     logger.d('Total range to query for: ${totalRangeToQueryFor.rangeSegments}\n'
         'Sub ranges in snapshots (DRIVE ID: $driveId): ${snapshotDriveHistory.subRanges.rangeSegments}\n'
         'Sub ranges in GQL (DRIVE ID: $driveId): ${gqlDriveHistorySubRanges.rangeSegments}');
@@ -770,40 +771,6 @@ class _SyncRepository implements SyncRepository {
         fileEntities.clear();
       }
 
-      final pool = Pool(_entityBatchSize);
-      final runningTasks = <Future<void>>[];
-
-      Future<void> parseTransaction(DriveEntityHistoryTransactionModel t) async {
-        await for (final entity in _arweave.streamEntitiesFromTransactions(
-          Stream.value(t),
-          driveKey,
-          driveId: drive.id,
-          ownerAddress: ownerAddress,
-          concurrency: 1,
-        )) {
-          if (entity is DriveEntity) {
-            driveEntities.add(entity);
-          } else if (entity is FolderEntity) {
-            folderEntities.add(entity);
-          } else if (entity is FileEntity) {
-            fileEntities.add(entity);
-          }
-        }
-
-        numberOfDriveEntitiesParsed++;
-
-        if (driveEntities.length +
-                folderEntities.length +
-                fileEntities.length >=
-            _entityBatchSize) {
-          await flush();
-        }
-
-        controller.add(
-          fetchPhaseWeight + driveEntityParseProgress() * parsePhaseWeight,
-        );
-      }
-
       await for (final t in transactions) {
         if (firstBlockHeight == null) {
           final block = t.transactionCommonMixin.block;
@@ -824,16 +791,36 @@ class _SyncRepository implements SyncRepository {
           controller.add(fetchPhasePercentage * fetchPhaseWeight);
         }
 
-        final future = pool.withResource(() => parseTransaction(t));
-        runningTasks.add(future);
-        future.whenComplete(() => runningTasks.remove(future));
+        await for (final entity in _arweave.streamEntitiesFromTransactions(
+          Stream.value(t),
+          driveKey,
+          driveId: drive.id,
+          ownerAddress: ownerAddress,
+          concurrency: _entityBatchSize,
+        )) {
+          if (entity is DriveEntity) {
+            driveEntities.add(entity);
+          } else if (entity is FolderEntity) {
+            folderEntities.add(entity);
+          } else if (entity is FileEntity) {
+            fileEntities.add(entity);
+          }
 
-        if (runningTasks.length >= _entityBatchSize) {
-          await Future.any(runningTasks);
+          if (driveEntities.length +
+                  folderEntities.length +
+                  fileEntities.length >=
+              _entityBatchSize) {
+            await flush();
+          }
         }
+
+        numberOfDriveEntitiesParsed++;
+
+        controller.add(
+          fetchPhaseWeight + driveEntityParseProgress() * parsePhaseWeight,
+        );
       }
 
-      await Future.wait(runningTasks);
       await flush();
 
       if (numberOfDriveEntitiesToParse == 0) {