feat(core): respect updates to M:N inverse sides and batch them (#4798)

Implements diffing for inverse sides of M:N relations (for SQL drivers).
This was previously working only if the items were initialized.

```ts
const tag = await em.findOne(BookTag, 1);
// tag.books in an inverse side
tag.books.add(em.getReference(Book, 123));
await em.flush();
```

The M:N updates are now also batched.

Closes #4564

packages/core/src/EntityManager.ts

@@ -1486,7 +1486,7 @@ export class EntityManager<D extends IDatabaseDriver = IDatabaseDriver> {
    * Tells the EntityManager to make an instance managed and persistent.
    * The entity will be entered into the database at or before transaction commit or as a result of the flush operation.
    */
-  persist<Entity extends object>(entity: Entity | Reference<Entity> | (Entity | Reference<Entity>)[]): this {
+  persist<Entity extends object>(entity: Entity | Reference<Entity> | Iterable<Entity | Reference<Entity>>): this {
     const em = this.getContext();
 
     if (Utils.isEntity(entity)) {
@@ -1515,7 +1515,7 @@ export class EntityManager<D extends IDatabaseDriver = IDatabaseDriver> {
    * Persists your entity immediately, flushing all not yet persisted changes to the database too.
    * Equivalent to `em.persist(e).flush()`.
    */
-  async persistAndFlush(entity: AnyEntity | Reference<AnyEntity> | (AnyEntity | Reference<AnyEntity>)[]): Promise<void> {
+  async persistAndFlush(entity: AnyEntity | Reference<AnyEntity> | Iterable<AnyEntity | Reference<AnyEntity>>): Promise<void> {
     await this.persist(entity).flush();
   }
 
@@ -1525,7 +1525,7 @@ export class EntityManager<D extends IDatabaseDriver = IDatabaseDriver> {
    *
    * To remove entities by condition, use `em.nativeDelete()`.
    */
-  remove<Entity extends object>(entity: Entity | Reference<Entity> | (Entity | Reference<Entity>)[]): this {
+  remove<Entity extends object>(entity: Entity | Reference<Entity> | Iterable<Entity | Reference<Entity>>): this {
     const em = this.getContext();
 
     if (Utils.isEntity<Entity>(entity)) {
@@ -1552,7 +1552,7 @@ export class EntityManager<D extends IDatabaseDriver = IDatabaseDriver> {
    * Removes an entity instance immediately, flushing all not yet persisted changes to the database too.
    * Equivalent to `em.remove(e).flush()`
    */
-  async removeAndFlush(entity: AnyEntity | Reference<AnyEntity>): Promise<void> {
+  async removeAndFlush(entity: AnyEntity | Reference<AnyEntity> | Iterable<AnyEntity | Reference<AnyEntity>>): Promise<void> {
     await this.remove(entity).flush();
   }
 

packages/core/src/drivers/DatabaseDriver.ts

@@ -92,10 +92,20 @@ export abstract class DatabaseDriver<C extends Connection> implements IDatabaseD
     throw new Error(`${this.constructor.name} does not use pivot tables`);
   }
 
-  async syncCollection<T extends object, O extends object>(coll: Collection<T, O>, options?: DriverMethodOptions): Promise<void> {
-    const pk = coll.property.targetMeta!.primaryKeys[0];
-    const data = { [coll.property.name]: coll.getIdentifiers(pk) } as EntityData<T>;
-    await this.nativeUpdate<T>(coll.owner.constructor.name, helper(coll.owner).getPrimaryKey() as FilterQuery<T>, data, options);
+  async syncCollections<T extends object, O extends object>(collections: Iterable<Collection<T, O>>, options?: DriverMethodOptions): Promise<void> {
+    for (const coll of collections) {
+      if (!coll.property.owner) {
+        if (coll.getSnapshot() === undefined) {
+          throw ValidationError.cannotModifyInverseCollection(coll.owner, coll.property);
+        }
+
+        continue;
+      }
+
+      const pk = coll.property.targetMeta!.primaryKeys[0];
+      const data = { [coll.property.name]: coll.getIdentifiers(pk) } as EntityData<T>;
+      await this.nativeUpdate<T>(coll.owner.constructor.name, helper(coll.owner).getPrimaryKey() as FilterQuery<T>, data, options);
+    }
   }
 
   mapResult<T extends object>(result: EntityDictionary<T>, meta?: EntityMetadata<T>, populate: PopulateOptions<T>[] = []): EntityData<T> | null {

packages/core/src/drivers/IDatabaseDriver.ts

@@ -51,7 +51,7 @@ export interface IDatabaseDriver<C extends Connection = Connection> {
 
   nativeDelete<T extends object>(entityName: string, where: FilterQuery<T>, options?: NativeDeleteOptions<T>): Promise<QueryResult<T>>;
 
-  syncCollection<T extends object, O extends object>(collection: Collection<T, O>, options?: DriverMethodOptions): Promise<void>;
+  syncCollections<T extends object, O extends object>(collections: Iterable<Collection<T, O>>, options?: DriverMethodOptions): Promise<void>;
 
   count<T extends object, P extends string = never>(entityName: string, where: FilterQuery<T>, options?: CountOptions<T, P>): Promise<number>;
 

packages/core/src/entity/ArrayCollection.ts

@@ -64,7 +64,7 @@ export class ArrayCollection<T extends object, O extends object> {
     }) as unknown as U[];
   }
 
-  add(entity: T | Reference<T> | (T | Reference<T>)[], ...entities: (T | Reference<T>)[]): void {
+  add(entity: T | Reference<T> | Iterable<T | Reference<T>>, ...entities: (T | Reference<T>)[]): void {
     entities = Utils.asArray(entity).concat(entities);
 
     for (const item of entities) {
@@ -79,12 +79,12 @@ export class ArrayCollection<T extends object, O extends object> {
     }
   }
 
-  set(items: (T | Reference<T>)[]): void {
-    if (this.compare(items.map(item => Reference.unwrapReference(item)))) {
+  set(items: Iterable<T | Reference<T>>): void {
+    if (this.compare(Utils.asArray(items).map(item => Reference.unwrapReference(item)))) {
       return;
     }
 
-    this.removeAll();
+    this.remove(this.items);
     this.add(items);
   }
 
@@ -123,7 +123,7 @@ export class ArrayCollection<T extends object, O extends object> {
    * is not the same as `em.remove()`. If we want to delete the entity by removing it from collection, we need to enable `orphanRemoval: true`,
    * which tells the ORM we don't want orphaned entities to exist, so we know those should be removed.
    */
-  remove(entity: T | Reference<T> | (T | Reference<T>)[], ...entities: (T | Reference<T>)[]): void {
+  remove(entity: T | Reference<T> | Iterable<T | Reference<T>>, ...entities: (T | Reference<T>)[]): void {
     entities = Utils.asArray(entity).concat(entities);
     let modified = false;
 

packages/core/src/entity/Collection.ts

@@ -158,21 +158,22 @@ export class Collection<T extends object, O extends object = object> extends Arr
     return super.toJSON() as unknown as EntityDTO<TT>[];
   }
 
-  override add<TT extends T>(entity: TT | Reference<TT> | (TT | Reference<TT>)[], ...entities: (TT | Reference<TT>)[]): void {
+  override add<TT extends T>(entity: TT | Reference<TT> | Iterable<TT | Reference<TT>>, ...entities: (TT | Reference<TT>)[]): void {
     entities = Utils.asArray(entity).concat(entities);
     const unwrapped = entities.map(i => Reference.unwrapReference(i)) as T[];
     unwrapped.forEach(entity => this.validateItemType(entity));
     this.modify('add', unwrapped);
     this.cancelOrphanRemoval(unwrapped);
   }
 
-  override set<TT extends T>(items: (TT | Reference<TT>)[]): void {
+  override set<TT extends T>(items: Iterable<TT | Reference<TT>>): void {
     if (!this.initialized) {
       this.initialized = true;
       this.snapshot = undefined;
     }
 
     super.set(items as T[]);
+    this.setDirty();
   }
 
   /**
@@ -187,7 +188,7 @@ export class Collection<T extends object, O extends object = object> extends Arr
   /**
    * @inheritDoc
    */
-  override remove<TT extends T>(entity: TT | Reference<TT> | (TT | Reference<TT>)[] | ((item: TT) => boolean), ...entities: (TT | Reference<TT>)[]): void {
+  override remove<TT extends T>(entity: TT | Reference<TT> | Iterable<TT | Reference<TT>> | ((item: TT) => boolean), ...entities: (TT | Reference<TT>)[]): void {
     if (entity instanceof Function) {
       for (const item of this.items) {
         if (entity(item as TT)) {
@@ -214,8 +215,7 @@ export class Collection<T extends object, O extends object = object> extends Arr
    * @inheritDoc
    */
   override removeAll(): void {
-    this.checkInitialized();
-    super.removeAll();
+    this.set([]);
   }
 
   override contains<TT extends T>(item: TT | Reference<TT>, check = true): boolean {

packages/core/src/unit-of-work/ChangeSetComputer.ts

@@ -155,7 +155,7 @@ export class ChangeSetComputer {
   }
 
   private processToMany<T extends object>(prop: EntityProperty<T>, changeSet: ChangeSet<T>): void {
-    const target = changeSet.entity[prop.name] as unknown as Collection<any>;
+    const target = changeSet.entity[prop.name] as Collection<any>;
 
     if (!target.isDirty()) {
       return;
@@ -169,8 +169,8 @@ export class ChangeSetComputer {
       }
     } else if (prop.kind === ReferenceKind.ONE_TO_MANY && target.getSnapshot() === undefined) {
       this.collectionUpdates.add(target);
-    } else {
-      target.setDirty(false); // inverse side with only populated items, nothing to persist
+    } else if (prop.kind === ReferenceKind.MANY_TO_MANY && !prop.owner) {
+      this.collectionUpdates.add(target);
     }
   }
 

packages/core/src/unit-of-work/UnitOfWork.ts

@@ -880,8 +880,9 @@ export class UnitOfWork {
     await this.commitExtraUpdates(ctx);
 
     // 6. collection updates
+    await this.em.getDriver().syncCollections(this.collectionUpdates, { ctx });
+
     for (const coll of this.collectionUpdates) {
-      await this.em.getDriver().syncCollection(coll, { ctx });
       coll.takeSnapshot();
     }
 

packages/core/src/utils/Utils.ts

@@ -399,16 +399,27 @@ export class Utils {
   /**
    * Normalize the argument to always be an array.
    */
-  static asArray<T>(data?: T | readonly T[], strict = false): T[] {
+  static asArray<T>(data?: T | readonly T[] | Iterable<T>, strict = false): T[] {
     if (typeof data === 'undefined' && !strict) {
       return [];
     }
 
-    if (data instanceof Set) {
+    if (this.isIterable(data)) {
       return Array.from(data);
     }
 
-    return Array.isArray(data!) ? data : [data as T];
+    return [data as T];
+  }
+
+  /**
+   * Checks if the value is iterable, but considers strings and buffers as not iterable.
+   */
+  static isIterable<T>(value: unknown): value is Iterable<T> {
+    if (value == null || typeof value === 'string' || ArrayBuffer.isView(value)) {
+      return false;
+    }
+
+    return typeof Object(value)[Symbol.iterator] === 'function';
   }
 
   /**