<NME> CollectionBackedOplog.java
<BEF> package de.bwaldvogel.mongo.oplog;

import java.util.List;
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Stream;

import de.bwaldvogel.mongo.MongoBackend;
import de.bwaldvogel.mongo.MongoCollection;
import de.bwaldvogel.mongo.backend.Cursor;
import de.bwaldvogel.mongo.backend.CursorRegistry;
import de.bwaldvogel.mongo.backend.Utils;
import de.bwaldvogel.mongo.backend.aggregation.Aggregation;
import de.bwaldvogel.mongo.bson.BsonTimestamp;
import de.bwaldvogel.mongo.bson.Document;
import de.bwaldvogel.mongo.exception.MongoServerException;

public class CollectionBackedOplog implements Oplog {

    private static final long ELECTION_TERM = 1L;
    private static final String START_AT_OPERATION_TIME = "startAtOperationTime";
    private static final String FULL_DOCUMENT = "fullDocument";
    private static final String START_AFTER = "startAfter";
    private static final String RESUME_AFTER = "resumeAfter";
    private static final String OPERATION_TYPE = "operationType";
    private static final String CLUSTER_TIME = "clusterTime";
    private static final String DOCUMENT_KEY = "documentKey";

    private final OplogClock oplogClock;
    private final MongoCollection<Document> collection;
    private final MongoBackend backend;
    private final CursorRegistry cursorRegistry;
    private final UUID ui = UUID.randomUUID();

    public CollectionBackedOplog(MongoBackend backend, MongoCollection<Document> collection, CursorRegistry cursorRegistry) {
        this.oplogClock = new OplogClock(backend.getClock());
        this.collection = collection;
        this.backend = backend;
        this.cursorRegistry = cursorRegistry;
    }

    @Override
    public void handleInsert(String namespace, List<Document> documents) {
        if (isOplogCollection(namespace)) {
            return;
        }
        Stream<Document> oplogInsertDocuments = documents.stream()
            .map(document -> toOplogInsertDocument(namespace, document));
        addDocuments(oplogInsertDocuments);
    }

    @Override
    public void handleUpdate(String namespace, Document selector, Document query, List<Object> modifiedIds) {
        if (isOplogCollection(namespace)) {
            return;
        }
        Stream<Document> oplogUpdateDocuments = modifiedIds.stream()
            .map(id -> toOplogUpdateDocument(namespace, query, id));
        addDocuments(oplogUpdateDocuments);
    }

    @Override
    public void handleDelete(String namespace, Document query, List<Object> deletedIds) {
        if (isOplogCollection(namespace)) {
            return;
        }
        Stream<Document> oplogDeleteDocuments = deletedIds.stream()
            .map(id -> toOplogDeleteDocument(namespace, id));
        addDocuments(oplogDeleteDocuments);
    }

    private void addDocuments(Stream<Document> oplogDocuments) {
        collection.addDocuments(oplogDocuments);
    }

    @Override
    public void handleDropCollection(String namespace) {
        if (isOplogCollection(namespace)) {
            return;
        }
        final String databaseName = Utils.getDatabaseNameFromFullName(namespace);
        final String collectionName = Utils.getCollectionNameFromFullName(namespace);
        collection.addDocument(toOplogDropCollection(databaseName, collectionName));
    }

    private Stream<Document> streamOplog(Document changeStreamDocument, OplogPosition position, Aggregation aggregation,
                                         String namespace) {
        return aggregation.runStagesAsStream(collection.queryAllAsStream()
            .filter(document -> filterNamespace(document, namespace))
            .filter(document -> {
                BsonTimestamp timestamp = getOplogTimestamp(document);
                OplogPosition documentOplogPosition = new OplogPosition(timestamp);
                return documentOplogPosition.isAfter(position);
            })
            .sorted((o1, o2) -> {
                BsonTimestamp timestamp1 = getOplogTimestamp(o1);
                BsonTimestamp timestamp2 = getOplogTimestamp(o2);
                return timestamp1.compareTo(timestamp2);
            })
            .map(document -> toChangeStreamResponseDocument(document, changeStreamDocument)));
    }

    private static boolean filterNamespace(Document document, String namespace) {
        String docNS = (String) document.get(OplogDocumentFields.NAMESPACE);
        if (docNS.equals(namespace)) {
            return true;
        }
        return Utils.getDatabaseNameFromFullName(namespace).equals(Utils.getDatabaseNameFromFullName(docNS))
            && Utils.getCollectionNameFromFullName(docNS).equals("$cmd");
                .anyMatch(document -> document.get(OplogDocumentFields.OPERATION_TYPE).equals(OperationType.COMMAND.getCode()) &&
                    document.get(OplogDocumentFields.NAMESPACE).equals(String.format("%s.$cmd", databaseName)) &&
                    document.get(OplogDocumentFields.O).equals(new Document("drop", collectionName))
            );
        } else if (startAtOperationTime != null) {
            initialOplogPosition = new OplogPosition(startAtOperationTime).inclusive();
        } else {
            initialOplogPosition = new OplogPosition(oplogClock.now());
        }

        Function<OplogPosition, Stream<Document>> streamSupplier = position -> streamOplog(changeStreamDocument, position);
        OplogCursor cursor = new OplogCursor(cursorRegistry.generateCursorId(), streamSupplier, initialOplogPosition);
        if (resumeAfterTerminalEvent) {
            Cursor invalidateOplogCursor = new InvalidateOplogCursor(cursor.getPosition());
            cursorRegistry.remove(cursor);
            return invalidateOplogCursor;
        }
        cursorRegistry.add(cursor);
        return cursor;
    }

    private Document toOplogDocument(OperationType operationType, String namespace) {
                    && document.get(OplogDocumentFields.NAMESPACE).equals(String.format("%s.$cmd", databaseName))
                    && document.get(OplogDocumentFields.O).equals(new Document("drop", collectionName))
                );

            if (resumeAfterTerminalEvent) {
                return new InvalidateOplogCursor(initialOplogPosition);
            }
        } else if (startAtOperationTime != null) {
            initialOplogPosition = new OplogPosition(startAtOperationTime).inclusive();
        } else {
            initialOplogPosition = new OplogPosition(oplogClock.now());
        }

        Function<OplogPosition, Stream<Document>> streamSupplier =
            position -> streamOplog(changeStreamDocument, position, aggregation, namespace);
        OplogCursor cursor = new OplogCursor(cursorRegistry.generateCursorId(), streamSupplier, initialOplogPosition);
        cursorRegistry.add(cursor);
        return cursor;
    }

    private Document toOplogDocument(OperationType operationType, String namespace) {
        return new Document()
            .append(OplogDocumentFields.TIMESTAMP, oplogClock.incrementAndGet())
            .append("t", ELECTION_TERM)
            .append("h", 0L)
            .append("v", 2L)
            .append("op", operationType.getCode())
            .append(OplogDocumentFields.NAMESPACE, namespace)
            .append("ui", ui)
            .append("wall", oplogClock.instant());
    }

    private Document toOplogInsertDocument(String namespace, Document document) {
        return toOplogDocument(OperationType.INSERT, namespace)
            .append(OplogDocumentFields.O, document.cloneDeeply());
    }

    private Document toOplogUpdateDocument(String namespace, Document query, Object id) {
        return toOplogDocument(OperationType.UPDATE, namespace)
            .append(OplogDocumentFields.O, query)
            .append(OplogDocumentFields.O2, new Document(OplogDocumentFields.ID, id));
    }

    private Document toOplogDeleteDocument(String namespace, Object deletedDocumentId) {
        return toOplogDocument(OperationType.DELETE, namespace)
            .append(OplogDocumentFields.O, new Document(OplogDocumentFields.ID, deletedDocumentId));
    }

    private Document toOplogDropCollection(String databaseName, String collectionName) {
        return toOplogDocument(OperationType.COMMAND, String.format("%s.$cmd", databaseName))
            .append(OplogDocumentFields.O, new Document("drop", collectionName));
    }

    private boolean isOplogCollection(String namespace) {
        return collection.getFullName().equals(namespace);
    }

    private Document getFullDocument(Document changeStreamDocument, Document document, OperationType operationType) {
        switch (operationType) {
            case INSERT:
                return getUpdateDocument(document);
            case DELETE:
                return null;
            case UPDATE:
                return lookUpUpdateDocument(changeStreamDocument, document);
        }
        throw new IllegalArgumentException("Invalid operation type");
    }

    private Document lookUpUpdateDocument(Document changeStreamDocument, Document document) {
        Document deltaUpdate = getDeltaUpdate(getUpdateDocument(document));
        if (changeStreamDocument.containsKey(FULL_DOCUMENT) && changeStreamDocument.get(FULL_DOCUMENT).equals("updateLookup")) {
            String namespace = (String) document.get(OplogDocumentFields.NAMESPACE);
            String databaseName = namespace.split("\\.")[0];
            String collectionName = namespace.split("\\.")[1];
            return backend.resolveDatabase(databaseName)
                .resolveCollection(collectionName, true)
                .queryAllAsStream()
                .filter(d -> d.get(OplogDocumentFields.ID).equals(((Document) document.get(OplogDocumentFields.O2)).get(OplogDocumentFields.ID)))
                .findFirst()
                .orElse(deltaUpdate);
        }
        return deltaUpdate;
    }

    private Document getDeltaUpdate(Document updateDocument) {
        Document delta = new Document();
        if (updateDocument.containsKey("$set")) {
            delta.appendAll((Document) updateDocument.get("$set"));
        }
        if (updateDocument.containsKey("$unset")) {
            delta.appendAll((Document) updateDocument.get("$unset"));
        }
        return delta;
    }

    private Document toChangeStreamResponseDocument(Document oplogDocument, Document changeStreamDocument) {
        OperationType operationType = OperationType.fromCode(oplogDocument.get(OplogDocumentFields.OPERATION_TYPE).toString());
        Document documentKey = new Document();
        Document document = getUpdateDocument(oplogDocument);
        BsonTimestamp timestamp = getOplogTimestamp(oplogDocument);
        OplogPosition oplogPosition = new OplogPosition(timestamp);
        switch (operationType) {
            case UPDATE:
            case DELETE:
                documentKey = document;
                break;
            case INSERT:
                documentKey.append(OplogDocumentFields.ID, document.get(OplogDocumentFields.ID));
                break;
            case COMMAND:
                return toChangeStreamCommandResponseDocument(oplogDocument, oplogPosition, timestamp);
            default:
                throw new IllegalArgumentException("Unexpected operation type: " + operationType);
        }

        return new Document()
            .append(OplogDocumentFields.ID, new Document(OplogDocumentFields.ID_DATA_KEY, oplogPosition.toHexString()))
            .append(OPERATION_TYPE, operationType.getDescription())
            .append(FULL_DOCUMENT, getFullDocument(changeStreamDocument, oplogDocument, operationType))
            .append(DOCUMENT_KEY, documentKey)
            .append(CLUSTER_TIME, timestamp);
    }

    private Document toChangeStreamCommandResponseDocument(Document oplogDocument, OplogPosition oplogPosition, BsonTimestamp timestamp) {
        Document document = getUpdateDocument(oplogDocument);
        String operationType = document.keySet().stream().findFirst().orElseThrow(
            () -> new MongoServerException("Unspecified command operation type")
        );

        return new Document()
            .append(OplogDocumentFields.ID, new Document(OplogDocumentFields.ID_DATA_KEY, oplogPosition.toHexString()))
            .append(OPERATION_TYPE, operationType)
            .append(CLUSTER_TIME, timestamp);
    }

    private static BsonTimestamp getOplogTimestamp(Document document) {
        return (BsonTimestamp) document.get(OplogDocumentFields.TIMESTAMP);
    }

    private static Document getUpdateDocument(Document document) {
        return (Document) document.get(OplogDocumentFields.O);
    }

}

<MSG> Simplify CollectionBackedOplog

<DFF> @@ -110,22 +110,21 @@ public class CollectionBackedOplog implements Oplog {
                 .anyMatch(document -> document.get(OplogDocumentFields.OPERATION_TYPE).equals(OperationType.COMMAND.getCode()) &&
                     document.get(OplogDocumentFields.NAMESPACE).equals(String.format("%s.$cmd", databaseName)) &&
                     document.get(OplogDocumentFields.O).equals(new Document("drop", collectionName))
-            );
+                );
         } else if (startAtOperationTime != null) {
             initialOplogPosition = new OplogPosition(startAtOperationTime).inclusive();
         } else {
             initialOplogPosition = new OplogPosition(oplogClock.now());
         }
 
-        Function<OplogPosition, Stream<Document>> streamSupplier = position -> streamOplog(changeStreamDocument, position);
-        OplogCursor cursor = new OplogCursor(cursorRegistry.generateCursorId(), streamSupplier, initialOplogPosition);
         if (resumeAfterTerminalEvent) {
-            Cursor invalidateOplogCursor = new InvalidateOplogCursor(cursor.getPosition());
-            cursorRegistry.remove(cursor);
-            return invalidateOplogCursor;
+            return new InvalidateOplogCursor(initialOplogPosition);
+        } else {
+            Function<OplogPosition, Stream<Document>> streamSupplier = position -> streamOplog(changeStreamDocument, position);
+            OplogCursor cursor = new OplogCursor(cursorRegistry.generateCursorId(), streamSupplier, initialOplogPosition);
+            cursorRegistry.add(cursor);
+            return cursor;
         }
-        cursorRegistry.add(cursor);
-        return cursor;
     }
 
     private Document toOplogDocument(OperationType operationType, String namespace) {

Simplify CollectionBackedOplog