Stream transactions directly into database

Author: atticusofsparta

lib/sync/domain/repositories/sync_repository.dart

@@ -187,8 +187,6 @@ class _SyncRepository implements SyncRepository {
             ? 0
             : _calculateSyncLastBlockHeight(drive.lastBlockHeight!),
         currentBlockHeight: currentBlockHeight,
-        transactionParseBatchSize:
-            200 ~/ (syncProgress.drivesCount - syncProgress.drivesSynced),
         ownerAddress: drive.ownerAddress,
         txFechedCallback: txFechedCallback,
       );
@@ -301,7 +299,6 @@ class _SyncRepository implements SyncRepository {
       ownerAddress: ownerAddress,
       lastBlockHeight: 0,
       currentBlockHeight: 0,
-      transactionParseBatchSize: 200,
       txFechedCallback: txFechedCallback,
     );
   }
@@ -538,7 +535,6 @@ class _SyncRepository implements SyncRepository {
     SecretKey? cipherKey,
     required int currentBlockHeight,
     required int lastBlockHeight,
-    required int transactionParseBatchSize,
     required String ownerAddress,
     Function(String driveId, int txCount)? txFechedCallback,
   }) async* {
@@ -629,9 +625,6 @@ class _SyncRepository implements SyncRepository {
         drive: drive,
         driveKey: driveKey?.key,
         currentBlockHeight: currentBlockHeight,
-        lastBlockHeight: lastBlockHeight,
-        batchSize: transactionParseBatchSize,
-        snapshotDriveHistory: snapshotDriveHistory,
         ownerAddress: ownerAddress,
       );
     } catch (e) {
@@ -727,10 +720,7 @@ class _SyncRepository implements SyncRepository {
     required Stream<DriveEntityHistoryTransactionModel> transactions,
     required Drive drive,
     required SecretKey? driveKey,
-    required int lastBlockHeight,
     required int currentBlockHeight,
-    required int batchSize,
-    required SnapshotDriveHistory snapshotDriveHistory,
     required String ownerAddress,
   }) {
     final controller = StreamController<double>();
@@ -743,8 +733,6 @@ class _SyncRepository implements SyncRepository {
           ? 0
           : numberOfDriveEntitiesParsed / numberOfDriveEntitiesToParse;
 
-      final batch = <DriveEntityHistoryTransactionModel>[];
-
       int? firstBlockHeight;
       late int totalBlockHeightDifference;
       var fetchPhasePercentage = 0.0;
@@ -769,7 +757,6 @@ class _SyncRepository implements SyncRepository {
           }
         }
 
-        batch.add(t);
         numberOfDriveEntitiesToParse++;
 
         if (firstBlockHeight != null) {
@@ -781,37 +768,15 @@ class _SyncRepository implements SyncRepository {
           controller.add(fetchPhasePercentage * fetchPhaseWeight);
         }
 
-        if (batch.length >= batchSize) {
-          await _processTransactionBatch(
-            batch,
-            drive,
-            driveKey,
-            lastBlockHeight,
-            currentBlockHeight,
-            batchSize,
-            ownerAddress,
-          );
-          numberOfDriveEntitiesParsed += batch.length;
-          controller.add(fetchPhaseWeight +
-              driveEntityParseProgress() * parsePhaseWeight);
-          batch.clear();
-        }
-      }
-
-      if (batch.isNotEmpty) {
-        await _processTransactionBatch(
-          batch,
+        await _processTransaction(
+          t,
           drive,
           driveKey,
-          lastBlockHeight,
-          currentBlockHeight,
-          batchSize,
           ownerAddress,
         );
-        numberOfDriveEntitiesParsed += batch.length;
+        numberOfDriveEntitiesParsed++;
         controller.add(fetchPhaseWeight +
             driveEntityParseProgress() * parsePhaseWeight);
-        batch.clear();
       }
 
       if (numberOfDriveEntitiesToParse == 0) {
@@ -836,22 +801,18 @@ class _SyncRepository implements SyncRepository {
     return controller.stream;
   }
 
-  Future<void> _processTransactionBatch(
-    List<DriveEntityHistoryTransactionModel> items,
+  Future<void> _processTransaction(
+    DriveEntityHistoryTransactionModel item,
     Drive drive,
     SecretKey? driveKey,
-    int lastBlockHeight,
-    int currentBlockHeight,
-    int batchSize,
     String ownerAddress,
   ) async {
     final driveEntities = <DriveEntity>[];
     final folderEntities = <FolderEntity>[];
     final fileEntities = <FileEntity>[];
-    var processed = 0;
 
     await for (final entity in _arweave.streamEntitiesFromTransactions(
-      items,
+      [item],
       driveKey,
       driveId: drive.id,
       ownerAddress: ownerAddress,
@@ -863,41 +824,20 @@ class _SyncRepository implements SyncRepository {
       } else if (entity is FileEntity) {
         fileEntities.add(entity);
       }
-
-      processed++;
-      if (processed >= batchSize) {
-        await _insertEntities(
-          drive,
-          driveEntities,
-          folderEntities,
-          fileEntities,
-        );
-        driveEntities.clear();
-        folderEntities.clear();
-        fileEntities.clear();
-        processed = 0;
-      }
     }
 
-    if (driveEntities.isNotEmpty ||
-        folderEntities.isNotEmpty ||
-        fileEntities.isNotEmpty) {
-      await _insertEntities(
-        drive,
-        driveEntities,
-        folderEntities,
-        fileEntities,
-      );
+    if (driveEntities.isEmpty &&
+        folderEntities.isEmpty &&
+        fileEntities.isEmpty) {
+      return;
     }
 
-    if (items.length < batchSize) {
-      await _driveDao.writeToDrive(DrivesCompanion(
-        id: Value(drive.id),
-        lastBlockHeight: Value(currentBlockHeight),
-        syncCursor: const Value(null),
-        isHidden: Value(drive.isHidden),
-      ));
-    }
+    await _insertEntities(
+      drive,
+      driveEntities,
+      folderEntities,
+      fileEntities,
+    );
   }
 
   Future<void> _insertEntities(

lib/sync/utils/batch_processor.dart

@@ -29,27 +29,4 @@ class BatchProcessor {
 
     list.clear();
   }
-
-  Stream<double> batchProcessStream<T>({
-    required Stream<T> stream,
-    required Stream<double> Function(List<T> items) endOfBatchCallback,
-    required int batchSize,
-  }) async* {
-    if (batchSize <= 0) {
-      throw ArgumentError('Batch size cannot be 0');
-    }
-
-    final buffer = <T>[];
-    await for (final item in stream) {
-      buffer.add(item);
-      if (buffer.length >= batchSize) {
-        yield* endOfBatchCallback(List<T>.from(buffer));
-        buffer.clear();
-      }
-    }
-
-    if (buffer.isNotEmpty) {
-      yield* endOfBatchCallback(List<T>.from(buffer));
-    }
-  }
 }

test/sync/utils/batch_processor_test.dart

@@ -1,5 +1,3 @@
-import 'dart:async';
-
 import 'package:ardrive/sync/utils/batch_processor.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:mocktail/mocktail.dart';
@@ -20,44 +18,6 @@ void main() {
       });
     });
 
-    group('batchProcessStream', () {
-      test('should produce no output for an empty stream', () {
-        final processor = BatchProcessor();
-        final controller = StreamController<int>();
-        controller.close();
-
-        expect(
-            processor.batchProcessStream<int>(
-                stream: controller.stream,
-                endOfBatchCallback: mockCallback.call,
-                batchSize: 5),
-            emitsDone);
-      });
-
-      test('should split stream into batches', () {
-        final processor = BatchProcessor();
-        final controller = StreamController<int>();
-
-        when(() => mockCallback(any())).thenAnswer((_) async* {
-          yield 1.0;
-        });
-
-        final stream = processor.batchProcessStream<int>(
-          stream: controller.stream,
-          endOfBatchCallback: mockCallback.call,
-          batchSize: 3,
-        );
-
-        controller.add(1);
-        controller.add(2);
-        controller.add(3);
-        controller.add(4);
-        controller.close();
-
-        expect(stream, emitsInOrder([1.0, 1.0, emitsDone]));
-      });
-    });
-
     test('should produce no output for an empty list', () {
       final processor = BatchProcessor();
       expect(