# OneToManyWithIndexedAssociationTableEngine.java

    package org.codefilarete.stalactite.engine.runtime.onetomany;

    import java.util.ArrayList;
    import java.util.Collection;
    import java.util.Collections;
    import java.util.List;
    import java.util.Set;
    import java.util.stream.Collectors;
    import java.util.stream.Stream;

    import org.codefilarete.reflection.Accessor;
    import org.codefilarete.stalactite.engine.diff.AbstractDiff;
    import org.codefilarete.stalactite.engine.diff.IndexedDiff;
    import org.codefilarete.stalactite.engine.listener.SelectListener;
    import org.codefilarete.stalactite.engine.runtime.AssociationRecordPersister;
    import org.codefilarete.stalactite.engine.runtime.CollectionUpdater;
    import org.codefilarete.stalactite.engine.runtime.ConfiguredRelationalPersister;
    import org.codefilarete.stalactite.engine.runtime.IndexedAssociationRecord;
    import org.codefilarete.stalactite.engine.runtime.IndexedAssociationRecordInsertionCascader;
    import org.codefilarete.stalactite.engine.runtime.IndexedAssociationTable;
    import org.codefilarete.stalactite.engine.runtime.load.AbstractJoinNode;
    import org.codefilarete.stalactite.engine.runtime.load.EntityJoinTree.JoinType;
    import org.codefilarete.stalactite.engine.runtime.load.EntityTreeInflater;
    import org.codefilarete.stalactite.engine.runtime.load.JoinNode;
    import org.codefilarete.stalactite.engine.runtime.onetomany.IndexedAssociationTableManyRelationDescriptor.InMemoryRelationHolder;
    import org.codefilarete.stalactite.engine.runtime.onetomany.OneToManyWithMappedAssociationEngine.AfterUpdateTrigger;
    import org.codefilarete.stalactite.mapping.EntityMapping;
    import org.codefilarete.stalactite.mapping.RowTransformer;
    import org.codefilarete.stalactite.query.api.Fromable;
    import org.codefilarete.stalactite.sql.Dialect;
    import org.codefilarete.stalactite.sql.ddl.structure.Column;
    import org.codefilarete.stalactite.sql.ddl.structure.Table;
    import org.codefilarete.stalactite.sql.result.ColumnedRow;
    import org.codefilarete.stalactite.sql.statement.WriteOperationFactory;
    import org.codefilarete.tool.Duo;
    import org.codefilarete.tool.collection.Iterables;
    import org.codefilarete.tool.collection.PairIterator;

    import static org.codefilarete.stalactite.engine.runtime.load.EntityJoinTree.ROOT_JOIN_NAME;
    import static org.codefilarete.tool.Nullable.nullable;
    import static org.codefilarete.tool.collection.Iterables.first;
    import static org.codefilarete.tool.collection.Iterables.minus;

    /**
     * @author Guillaume Mary
     */
    public class OneToManyWithIndexedAssociationTableEngine<
    		SRC,
    		TRGT,
    		SRCID,
    		TRGTID,
    		S extends Collection<TRGT>,
    		LEFTTABLE extends Table<LEFTTABLE>,
    		RIGHTTABLE extends Table<RIGHTTABLE>,
    		ASSOCIATIONTABLE extends IndexedAssociationTable<ASSOCIATIONTABLE, LEFTTABLE, RIGHTTABLE, SRCID, TRGTID>>
    		extends OneToManyWithAssociationTableEngine<SRC, TRGT, SRCID, TRGTID, S, IndexedAssociationRecord, ASSOCIATIONTABLE> {

    	/** Column that stores index value */
    	private final Column<ASSOCIATIONTABLE, Integer> indexColumn;

    	public OneToManyWithIndexedAssociationTableEngine(ConfiguredRelationalPersister<SRC, SRCID> sourcePersister,
    													  ConfiguredRelationalPersister<TRGT, TRGTID> targetPersister,
    													  IndexedAssociationTableManyRelationDescriptor<SRC, TRGT, S, SRCID> manyRelationDescriptor,
    													  AssociationRecordPersister<IndexedAssociationRecord, ASSOCIATIONTABLE> associationPersister,
    													  Column<ASSOCIATIONTABLE, Integer> indexColumn,
    													  WriteOperationFactory writeOperationFactory,
    													  Dialect dialect) {
    		super(sourcePersister, targetPersister, manyRelationDescriptor, associationPersister, writeOperationFactory, dialect);
    		this.indexColumn = indexColumn;
    	}

    	@Override
    	public IndexedAssociationTableManyRelationDescriptor<SRC, TRGT, S, SRCID> getManyRelationDescriptor() {
    		return (IndexedAssociationTableManyRelationDescriptor<SRC, TRGT, S, SRCID>) super.getManyRelationDescriptor();
    	}

    	@Override
    	public String addSelectCascade(boolean loadSeparately) {
    		addIndexSelection(loadSeparately);

    		// We trigger subgraph load event (via targetSelectListener) on loading of our graph.
    		// Done for instance for event consumers that initialize some things, because given ids of methods are those of source entity
    		SelectListener<TRGT, TRGTID> targetSelectListener = targetPersister.getPersisterListener().getSelectListener();
    		sourcePersister.addSelectListener(new SelectListener<SRC, SRCID>() {
    			@Override
    			public void beforeSelect(Iterable<SRCID> ids) {
    				// since ids are not those of its entities, we should not pass them as argument, this will only initialize things if needed
    				targetSelectListener.beforeSelect(Collections.emptyList());
    			}

    			@Override
    			public void afterSelect(Set<? extends SRC> result) {
    				Set<TRGT> collect = Iterables.stream(result).flatMap(src -> nullable(manyRelationDescriptor.getCollectionAccessPoint().get(src))
    						.map(Collection::stream)
    						.getOr(Stream.empty()))
    						.collect(Collectors.toSet());
    				targetSelectListener.afterSelect(collect);
    			}

    			@Override
    			public void onSelectError(Iterable<SRCID> ids, RuntimeException exception) {
    				// since ids are not those of its entities, we should not pass them as argument
    				targetSelectListener.onSelectError(Collections.emptyList(), exception);
    			}
    		});
    		return null;
    	}

    	private void addIndexSelection(boolean loadSeparately) {
    		// we join on the association table and add bean association in memory
    		String associationTableJoinNodeName = sourcePersister.getEntityJoinTree().addPassiveJoin(ROOT_JOIN_NAME,
    				associationPersister.getMainTable().getOneSideKey(),
    				associationPersister.getMainTable().getOneSideForeignKey(),
    				JoinType.OUTER,
    				// we must add all the columns to make them available while decoding the row to create an IndexedAssociationRecord
    				associationPersister.getMainTable().getColumns());

    		// we add target subgraph joins to main persister
    		String rightEntityJoinName = targetPersister.joinAsMany(associationTableJoinNodeName, sourcePersister, manyRelationDescriptor.getCollectionAccessPoint(),
    				associationPersister.getMainTable().getManySideForeignKey(), associationPersister.getMainTable().getManySideKey(),
    				manyRelationDescriptor.getRelationFixer(), columnedRow -> {
    					TRGTID identifier = targetPersister.getMapping().getIdMapping().getIdentifierAssembler().assemble(columnedRow);
    					// indexColumn column value is took on join of association table, not target table, so we have to grab it
    					JoinNode<IndexedAssociationRecord, Fromable> join = (JoinNode<IndexedAssociationRecord, Fromable>) sourcePersister.getEntityJoinTree().getJoin(associationTableJoinNodeName);
    					ColumnedRow rowDecoder = EntityTreeInflater.currentContext().getDecoder(join);
    					Integer targetEntityIndex = rowDecoder.get(indexColumn);
    					return identifier + "-" + targetEntityIndex;
    				}, true, loadSeparately);

    		// Implementation note: we keep the object indexes and put the sorted entities in a temporary Collection, then add them all to the target List
    		RowTransformer<IndexedAssociationRecord> associationRecordProvider = associationPersister.getMapping().getRowTransformer();
    		sourcePersister.addSelectListener(new SelectListener<SRC, SRCID>() {
    			@Override
    			public void beforeSelect(Iterable<SRCID> ids) {
    				getRelationFixer().init();
    			}

    			@Override
    			public void afterSelect(Set<? extends SRC> result) {
    				getRelationFixer().applySort(result);
    				cleanContext();
    			}

    			@Override
    			public void onSelectError(Iterable<SRCID> ids, RuntimeException exception) {
    				cleanContext();
    			}

    			private void cleanContext() {
    				getRelationFixer().clear();
    			}
    		});
    		AbstractJoinNode<TRGT, Fromable, Fromable, TRGTID> join = (AbstractJoinNode<TRGT, Fromable, Fromable, TRGTID>) sourcePersister.getEntityJoinTree().getJoin(rightEntityJoinName);
    		JoinNode<IndexedAssociationRecord, Fromable> associationTableJoinNode = (JoinNode<IndexedAssociationRecord, Fromable>) sourcePersister.getEntityJoinTree().getJoin(associationTableJoinNodeName);
    		join.setConsumptionListener((trgt, columnValueProvider) -> {
    			IndexedAssociationRecord associationRecord = associationRecordProvider.transform(EntityTreeInflater.currentContext().getDecoder(associationTableJoinNode));
    			((InMemoryRelationHolder) manyRelationDescriptor.getRelationFixer()).addIndex(associationRecord.getLeft(), trgt, associationRecord.getIndex());
    		});
    	}

    	private InMemoryRelationHolder<SRC, SRCID, TRGT, S> getRelationFixer() {
    		return getManyRelationDescriptor().getRelationFixer();
    	}

    	@Override
    	public void addUpdateCascade(ConfiguredRelationalPersister<TRGT, TRGTID> targetPersister) {

    		// NB: we don't have any reverseSetter (for applying source entity to reverse side (target entity)), because this is only relevant
    		// when association is mapped without intermediary table (owned by "many-side" entity)
    		CollectionUpdater<SRC, TRGT, S> collectionUpdater = new CollectionUpdater<SRC, TRGT, S>(manyRelationDescriptor.getCollectionAccessPoint(), targetPersister, null, getManyRelationDescriptor().isOrphanRemoval()) {

    			@Override
    			protected AssociationTableUpdateContext newUpdateContext(Duo<SRC, SRC> updatePayload) {
    				return new AssociationTableUpdateContext(updatePayload);
    			}

    			@Override
    			protected void onHeldElements(UpdateContext updateContext, AbstractDiff<TRGT> diff) {
    				super.onHeldElements(updateContext, diff);
    				IndexedDiff indexedDiff = (IndexedDiff) diff;
    				Set<Integer> minus = minus(indexedDiff.getReplacerIndexes(), indexedDiff.getSourceIndexes());
    				Integer index = first(minus);
    				if (index != null ) {
    					SRC leftIdentifier = updateContext.getPayload().getLeft();
    					PairIterator<Integer, Integer> diffIndexIterator = new PairIterator<>(indexedDiff.getReplacerIndexes(), indexedDiff.getSourceIndexes());
    					diffIndexIterator.forEachRemaining(d -> {
    						if (!d.getLeft().equals(d.getRight()))
    							((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeUpdated().add(new Duo<>(
    									newRecord(leftIdentifier, diff.getSourceInstance(), d.getLeft()),
    									newRecord(leftIdentifier, diff.getSourceInstance(), d.getRight())));
    					});
    				}
    			}

    			@Override
    			protected Set<? extends AbstractDiff<TRGT>> diff(Collection<TRGT> modified, Collection<TRGT> unmodified) {
    				return getDiffer().diffOrdered(unmodified, modified);
    			}

    			@Override
    			protected void onAddedElements(UpdateContext updateContext, AbstractDiff<TRGT> diff) {
    				super.onAddedElements(updateContext, diff);
    				SRC leftIdentifier = updateContext.getPayload().getLeft();
    				((IndexedDiff<TRGT>) diff).getReplacerIndexes().forEach(idx ->
    						((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeInserted().add(
    								newRecord(leftIdentifier, diff.getReplacingInstance(), idx)));
    			}

    			@Override
    			protected void onRemovedElements(UpdateContext updateContext, AbstractDiff<TRGT> diff) {
    				super.onRemovedElements(updateContext, diff);
    				SRC leftIdentifier = updateContext.getPayload().getLeft();
    				((IndexedDiff<TRGT>) diff).getSourceIndexes().forEach(idx ->
    						((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeDeleted().add(
    								newRecord(leftIdentifier, diff.getSourceInstance(), idx)));
    			}

    			@Override
    			protected void updateTargets(UpdateContext updateContext, boolean allColumnsStatement) {
    				super.updateTargets(updateContext, allColumnsStatement);
    				// association records can't be updated because they are primary key elements for us (see EmbeddedClassMapping and its updatableProperties computation),
    				// so we ask for their deletion + creation
    				List<Duo<IndexedAssociationRecord, IndexedAssociationRecord>> associationRecordsToBeUpdated = ((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeUpdated();
    				associationPersister.delete(Iterables.collectToList(associationRecordsToBeUpdated, Duo::getRight));
    				associationPersister.insert(Iterables.collectToList(associationRecordsToBeUpdated, Duo::getLeft));
    			}

    			@Override
    			protected void insertTargets(UpdateContext updateContext) {
    				// we insert targets before association records to satisfy integrity constraint
    				super.insertTargets(updateContext);
    				associationPersister.insert(((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeInserted());

    			}

    			@Override
    			protected void deleteTargets(UpdateContext updateContext) {
    				// we delete association records before targets to satisfy integrity constraint
    				associationPersister.delete(((AssociationTableUpdateContext) updateContext).getAssociationRecordsToBeDeleted());
    				super.deleteTargets(updateContext);
    			}

    			class AssociationTableUpdateContext extends UpdateContext {

    				private final List<IndexedAssociationRecord> associationRecordsToBeInserted = new ArrayList<>();
    				private final List<IndexedAssociationRecord> associationRecordsToBeDeleted = new ArrayList<>();
    				private final List<Duo<IndexedAssociationRecord, IndexedAssociationRecord>> associationRecordsToBeUpdated = new ArrayList<>();

    				public AssociationTableUpdateContext(Duo<SRC, SRC> updatePayload) {
    					super(updatePayload);
    				}

    				public List<IndexedAssociationRecord> getAssociationRecordsToBeInserted() {
    					return associationRecordsToBeInserted;
    				}

    				public List<IndexedAssociationRecord> getAssociationRecordsToBeDeleted() {
    					return associationRecordsToBeDeleted;
    				}

    				public List<Duo<IndexedAssociationRecord, IndexedAssociationRecord>> getAssociationRecordsToBeUpdated() {
    					return associationRecordsToBeUpdated;
    				}
    			}
    		};

    		// Can we cascade update on target entities ? it depends on relation maintenance mode
    		if (!getManyRelationDescriptor().isMaintainAssociationOnly()) {
    			persisterListener.addUpdateListener(new AfterUpdateTrigger<>(collectionUpdater));
    		}
    	}

    	@Override
    	protected IndexedAssociationRecordInsertionCascader<SRC, TRGT, SRCID, TRGTID, S> newRecordInsertionCascader(
    			Accessor<SRC, S> collectionGetter,
    			AssociationRecordPersister<IndexedAssociationRecord, ASSOCIATIONTABLE> associationPersister,
    			EntityMapping<SRC, SRCID, ?> mappingStrategy,
    			EntityMapping<TRGT, TRGTID, ?> targetStrategy) {
    		return new IndexedAssociationRecordInsertionCascader<>(associationPersister, collectionGetter, mappingStrategy, targetStrategy);
    	}

    	@Override
    	protected IndexedAssociationRecord newRecord(SRC e, TRGT target, int index) {
    		return new IndexedAssociationRecord(sourcePersister.getMapping().getId(e), targetPersister.getMapping().getId(target), index);
    	}
    }