Handle cascading inserts (triggered by the deletion of entries via on…

…Conflict:replace)

floor/lib/src/adapter/insertion_adapter.dart

@@ -8,18 +8,18 @@ class InsertionAdapter<T> {
   final DatabaseExecutor _database;
   final String _entityName;
   final Map<String, Object?> Function(T) _valueMapper;
-  final StreamController<Set<String>>? _changeListener;
+  final void Function(bool) _changeHandler;
 
   InsertionAdapter(
     final DatabaseExecutor database,
     final String entityName,
     final Map<String, Object?> Function(T) valueMapper, [
-    final StreamController<Set<String>>? changeListener,
+    final void Function(bool)? changeHandler,
   ])  : assert(entityName.isNotEmpty),
         _database = database,
         _entityName = entityName,
         _valueMapper = valueMapper,
-        _changeListener = changeListener;
+        _changeHandler = changeHandler ?? ((bool isReplace) {/* do nothing */});
 
   Future<void> insert(
     final T item,
@@ -42,7 +42,7 @@ class InsertionAdapter<T> {
       );
     }
     await batch.commit(noResult: true);
-    _changeListener?.add({_entityName});
+    _changeHandler(onConflictStrategy == OnConflictStrategy.replace);
   }
 
   Future<int> insertAndReturnId(
@@ -66,7 +66,8 @@ class InsertionAdapter<T> {
       );
     }
     final result = (await batch.commit(noResult: false)).cast<int>();
-    if (result.isNotEmpty) _changeListener?.add({_entityName});
+    if (result.isNotEmpty)
+      _changeHandler(onConflictStrategy == OnConflictStrategy.replace);
     return result;
   }
 
@@ -79,7 +80,8 @@ class InsertionAdapter<T> {
       _valueMapper(item),
       conflictAlgorithm: onConflictStrategy.asSqfliteConflictAlgorithm(),
     );
-    if (result != 0) _changeListener?.add({_entityName});
+    if (result != 0)
+      _changeHandler(onConflictStrategy == OnConflictStrategy.replace);
     return result;
   }
 }

floor/test/adapter/insertion_adapter_test.dart

@@ -182,11 +182,14 @@ void main() {
     // ignore: close_sinks
     final mockStreamController = MockStreamController<Set<String>>();
 
+    final changeHandler = (bool isReplace) {
+      mockStreamController.add({entityName});
+    };
     final underTest = InsertionAdapter(
       mockDatabaseExecutor,
       entityName,
       valueMapper,
-      mockStreamController,
+      changeHandler,
     );
 
     tearDown(() {

floor/test/integration/stream_query_test.dart

@@ -137,13 +137,20 @@ void main() {
           actual,
           emitsInOrder(<List<Dog>>[
             [], // initial state,
+            [], // after inserting person1, [1]
+            [], // after inserting person2, [1]
             [dog1], // after inserting dog1
             [dog1], // after inserting dog2
             [], // after removing person1, which triggers cascade remove
           ]));
+      // [1] due to insert method having onConflict:replace, dog entries could be affected by this query, so a stream event is triggered.
 
       await personDao.insertPerson(person1);
+      // avoid that delete happens before the re-execution of
+      // the select query for the stream
+      await Future<void>.delayed(const Duration(milliseconds: 100));
       await personDao.insertPerson(person2);
+      await Future<void>.delayed(const Duration(milliseconds: 100));
 
       await database.dogDao.insertDog(dog1);
 

floor_generator/lib/writer/dao_writer.dart

@@ -1,5 +1,6 @@
 import 'package:code_builder/code_builder.dart';
 import 'package:floor_generator/misc/extension/string_extension.dart';
+import 'package:floor_generator/misc/extension/iterable_extension.dart';
 import 'package:floor_generator/value_object/dao.dart';
 import 'package:floor_generator/value_object/deletion_method.dart';
 import 'package:floor_generator/value_object/entity.dart';
@@ -83,12 +84,15 @@ class DaoWriter extends Writer {
         final valueMapper =
             '(${entity.classElement.displayName} item) => ${entity.valueMapping}';
 
-        final requiresChangeListener =
-            dbHasViewStreams || streamEntities.contains(entity);
+        // create a special change handler which decides case-by-case:
+        // if the insertion happens with onConflict:replace, consider the insertion like a deletion.
+        // if it will not replace (e.g. abort or ignore), only output the single entity at most.
+        final changeHandler = _generateChangeHandler(
+            foreignKeyRelationships.getAffectedByDelete(entity), entity);
 
         constructorBuilder
           ..initializers.add(Code(
-              "$fieldName = InsertionAdapter(database, '${entity.name}', $valueMapper${requiresChangeListener ? ', changeListener' : ''})"));
+              "$fieldName = InsertionAdapter(database, '${entity.name}', $valueMapper$changeHandler)"));
       }
     }
 
@@ -221,15 +225,30 @@ class DaoWriter extends Writer {
   ///
   /// The affected set can be generated with [getAffectedByUpdateEntities]
   /// and [getAffectedByDeleteEntities]
-  String _generateChangeHandler(final Set<Entity> affected) {
+  String _generateChangeHandler(final Set<Entity> affected,
+      [Entity? insertionEntity]) {
     final toNotify = streamEntities.intersection(affected);
 
-    if (toNotify.isNotEmpty || dbHasViewStreams)
+    if (toNotify.isNotEmpty || dbHasViewStreams) {
       // if there are streaming views, create a new handler even if the set
       // is empty. This will only trigger a reload of the views.
-      return ', () => changeListener.add(const {${toNotify.map((e) => e.name.toLiteral()).join(', ')}})';
-    else
+      final set = toNotify.map((e) => e.name).toSetLiteral();
+      if (insertionEntity == null) {
+        return ', () => changeListener.add($set)';
+      } else {
+        final singleSet = (streamEntities.contains(insertionEntity)
+                ? {insertionEntity.name}
+                : <String>{})
+            .toSetLiteral();
+        if (singleSet == set) {
+          return ', (isReplace) => changeListener.add($set)';
+        } else {
+          return ', (isReplace) => changeListener.add(isReplace?$set:$singleSet)';
+        }
+      }
+    } else {
       // do not generate a Handler if the listener doesn't have to be updated
       return '';
+    }
   }
 }

floor_generator/test/writer/dao_writer_test.dart

@@ -8,6 +8,7 @@ import 'package:floor_generator/processor/entity_processor.dart';
 import 'package:floor_generator/processor/view_processor.dart';
 import 'package:floor_generator/value_object/dao.dart';
 import 'package:floor_generator/value_object/entity.dart';
+import 'package:floor_generator/value_object/foreign_key.dart';
 import 'package:floor_generator/value_object/primary_key.dart';
 import 'package:floor_generator/writer/dao_writer.dart';
 import 'package:source_gen/source_gen.dart';
@@ -129,7 +130,7 @@ void main() {
                     'Person',
                     (Person item) =>
                         <String, Object?>{'id': item.id, 'name': item.name},
-                    changeListener),
+                    (isReplace) => changeListener.add(const {'Person'})),
                 _personUpdateAdapter = UpdateAdapter(
                     database,
                     'Person',
@@ -180,6 +181,76 @@ void main() {
       '''));
   });
 
+  test('create DAO stream query with onconflict:replace insert', () async {
+    // add a simulated entity which should be affected if a Person is replaced
+    final dogEntity = Entity(
+        FakeClassElement(),
+        'Dog',
+        [],
+        PrimaryKey([], false),
+        [
+          ForeignKey(
+              'Person',
+              ['id'],
+              ['owner_id'],
+              annotations.ForeignKeyAction.noAction,
+              annotations.ForeignKeyAction.setDefault)
+        ],
+        [],
+        false,
+        '',
+        '',
+        null);
+
+    final dao = await _createDao('''
+        @dao
+        abstract class PersonDao {
+          @Query('SELECT * FROM person')
+          Stream<List<Person>> findAllPersonsAsStream();
+          
+          @Insert(onConflict: OnConflictStrategy.replace)
+          Future<void> insertPerson(Person person);
+        }
+      ''', dogEntity);
+
+    final streamEntities = {...dao.streamEntities, dogEntity};
+
+    final actual = DaoWriter(dao, streamEntities, dao.streamViews.isNotEmpty,
+            ForeignKeyMap.fromEntities(streamEntities))
+        .write();
+
+    expect(actual, equalsDart(r'''
+        class _$PersonDao extends PersonDao {
+          _$PersonDao(this.database, this.changeListener)
+              : _queryAdapter = QueryAdapter(database, changeListener),
+                _personInsertionAdapter = InsertionAdapter(
+                    database,
+                    'Person',
+                    (Person item) =>
+                        <String, Object?>{'id': item.id, 'name': item.name},
+                    (isReplace) => changeListener.add(isReplace ? const {'Person','Dog'}: const {'Person'}));
+        
+          final sqflite.DatabaseExecutor database;
+        
+          final StreamController<Set<String>> changeListener;
+        
+          final QueryAdapter _queryAdapter;
+        
+          final InsertionAdapter<Person> _personInsertionAdapter;
+        
+          @override
+          Stream<List<Person>> findAllPersonsAsStream() {
+            return _queryAdapter.queryListStream('SELECT * FROM person', mapper: (Map<String, Object?> row) => Person(row['id'] as int, row['name'] as String), queryableName: 'Person', isView: false);
+          }
+          
+          @override
+          Future<void> insertPerson(Person person) async {
+            await _personInsertionAdapter.insert(person, OnConflictStrategy.replace);
+          }
+        }
+      '''));
+  });
+
   test('create DAO aware of other entity stream query', () async {
     final dao = await _createDao('''
         @dao
@@ -207,7 +278,7 @@ void main() {
                     'Person',
                     (Person item) =>
                         <String, Object?>{'id': item.id, 'name': item.name},
-                    changeListener),
+                    (isReplace) => changeListener.add(const {'Person'})),
                 _personUpdateAdapter = UpdateAdapter(
                     database,
                     'Person',
@@ -355,7 +426,7 @@ void main() {
                     'Person',
                     (Person item) =>
                         <String, Object?>{'id': item.id, 'name': item.name},
-                    changeListener),
+                    (isReplace) => changeListener.add(const {})),                
                 _personUpdateAdapter = UpdateAdapter(
                     database,
                     'Person',
@@ -400,7 +471,8 @@ void main() {
   });
 }
 
-Future<Dao> _createDao(final String dao) async {
+Future<Dao> _createDao(final String dao,
+    [final Entity? additionalEntity]) async {
   final library = await resolveSource('''
       library test;
       
@@ -436,6 +508,10 @@ Future<Dao> _createDao(final String dao) async {
       .map((classElement) => EntityProcessor(classElement, {}).process())
       .toList();
 
+  if (additionalEntity != null) {
+    entities.add(additionalEntity);
+  }
+
   final views = library.classes
       .where((classElement) =>
           classElement.hasAnnotation(annotations.DatabaseView))