"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.ChangeStream = void 0;
const util_1 = require("util");
const collection_1 = require("./collection");
const constants_1 = require("./constants");
const change_stream_cursor_1 = require("./cursor/change_stream_cursor");
const db_1 = require("./db");
const error_1 = require("./error");
const mongo_client_1 = require("./mongo_client");
const mongo_types_1 = require("./mongo_types");
const utils_1 = require("./utils");
/** @internal */
const kCursorStream = Symbol('cursorStream');
/** @internal */
const kClosed = Symbol('closed');
/** @internal */
const kMode = Symbol('mode');
const CHANGE_STREAM_OPTIONS = [
    'resumeAfter',
    'startAfter',
    'startAtOperationTime',
    'fullDocument',
    'fullDocumentBeforeChange',
    'showExpandedEvents'
];
const CHANGE_DOMAIN_TYPES = {
    COLLECTION: Symbol('Collection'),
    DATABASE: Symbol('Database'),
    CLUSTER: Symbol('Cluster')
};
const CHANGE_STREAM_EVENTS = [constants_1.RESUME_TOKEN_CHANGED, constants_1.END, constants_1.CLOSE];
const NO_RESUME_TOKEN_ERROR = 'A change stream document has been received that lacks a resume token (_id).';
const CHANGESTREAM_CLOSED_ERROR = 'ChangeStream is closed';
/**
 * Creates a new Change Stream instance. Normally created using {@link Collection#watch|Collection.watch()}.
 * @public
 */
class ChangeStream extends mongo_types_1.TypedEventEmitter {
    /**
     * @internal
     *
     * @param parent - The parent object that created this change stream
     * @param pipeline - An array of {@link https://docs.mongodb.com/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents
     */
    constructor(parent, pipeline = [], options = {}) {
        super();
        /**
         * @internal
         *
         * TODO(NODE-4320): promisify selectServer and refactor this code to be async
         *
         * we promisify _processErrorIteratorModeCallback until we have a promisifed version of selectServer.
         */
        // eslint-disable-next-line @typescript-eslint/unbound-method
        this._processErrorIteratorMode = (0, util_1.promisify)(this._processErrorIteratorModeCallback);
        this.pipeline = pipeline;
        this.options = options;
        if (parent instanceof collection_1.Collection) {
            this.type = CHANGE_DOMAIN_TYPES.COLLECTION;
        }
        else if (parent instanceof db_1.Db) {
            this.type = CHANGE_DOMAIN_TYPES.DATABASE;
        }
        else if (parent instanceof mongo_client_1.MongoClient) {
            this.type = CHANGE_DOMAIN_TYPES.CLUSTER;
        }
        else {
            throw new error_1.MongoChangeStreamError('Parent provided to ChangeStream constructor must be an instance of Collection, Db, or MongoClient');
        }
        this.parent = parent;
        this.namespace = parent.s.namespace;
        if (!this.options.readPreference && parent.readPreference) {
            this.options.readPreference = parent.readPreference;
        }
        // Create contained Change Stream cursor
        this.cursor = this._createChangeStreamCursor(options);
        this[kClosed] = false;
        this[kMode] = false;
        // Listen for any `change` listeners being added to ChangeStream
        this.on('newListener', eventName => {
            if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {
                this._streamEvents(this.cursor);
            }
        });
        this.on('removeListener', eventName => {
            var _a;
            if (eventName === 'change' && this.listenerCount('change') === 0 && this.cursor) {
                (_a = this[kCursorStream]) === null || _a === void 0 ? void 0 : _a.removeAllListeners('data');
            }
        });
    }
    /** @internal */
    get cursorStream() {
        return this[kCursorStream];
    }
    /** The cached resume token that is used to resume after the most recently returned change. */
    get resumeToken() {
        var _a;
        return (_a = this.cursor) === null || _a === void 0 ? void 0 : _a.resumeToken;
    }
    hasNext(callback) {
        this._setIsIterator();
        // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
        // Shadowing is intentional here.  We want to override the `callback` variable
        // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
        return (0, utils_1.maybePromise)(callback, callback => {
            (async () => {
                try {
                    const hasNext = await this.cursor.hasNext();
                    return hasNext;
                }
                catch (error) {
                    try {
                        await this._processErrorIteratorMode(error);
                        const hasNext = await this.cursor.hasNext();
                        return hasNext;
                    }
                    catch (error) {
                        await this.close().catch(err => err);
                        throw error;
                    }
                }
            })().then(hasNext => callback(undefined, hasNext), error => callback(error));
        });
    }
    next(callback) {
        this._setIsIterator();
        // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
        // Shadowing is intentional here.  We want to override the `callback` variable
        // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
        return (0, utils_1.maybePromise)(callback, callback => {
            (async () => {
                try {
                    const change = await this.cursor.next();
                    const processedChange = this._processChange(change !== null && change !== void 0 ? change : null);
                    return processedChange;
                }
                catch (error) {
                    try {
                        await this._processErrorIteratorMode(error);
                        const change = await this.cursor.next();
                        const processedChange = this._processChange(change !== null && change !== void 0 ? change : null);
                        return processedChange;
                    }
                    catch (error) {
                        await this.close().catch(err => err);
                        throw error;
                    }
                }
            })().then(change => callback(undefined, change), error => callback(error));
        });
    }
    tryNext(callback) {
        this._setIsIterator();
        // TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
        // Shadowing is intentional here.  We want to override the `callback` variable
        // from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
        return (0, utils_1.maybePromise)(callback, callback => {
            (async () => {
                try {
                    const change = await this.cursor.tryNext();
                    return change !== null && change !== void 0 ? change : null;
                }
                catch (error) {
                    try {
                        await this._processErrorIteratorMode(error);
                        const change = await this.cursor.tryNext();
                        return change !== null && change !== void 0 ? change : null;
                    }
                    catch (error) {
                        await this.close().catch(err => err);
                        throw error;
                    }
                }
            })().then(change => callback(undefined, change), error => callback(error));
        });
    }
    /** Is the cursor closed */
    get closed() {
        return this[kClosed] || this.cursor.closed;
    }
    close(callback) {
        this[kClosed] = true;
        return (0, utils_1.maybePromise)(callback, cb => {
            const cursor = this.cursor;
            return cursor.close(err => {
                this._endStream();
                return cb(err);
            });
        });
    }
    /**
     * Return a modified Readable stream including a possible transform method.
     *
     * NOTE: When using a Stream to process change stream events, the stream will
     * NOT automatically resume in the case a resumable error is encountered.
     *
     * @throws MongoChangeStreamError if the underlying cursor or the change stream is closed
     */
    stream(options) {
        if (this.closed) {
            throw new error_1.MongoChangeStreamError(CHANGESTREAM_CLOSED_ERROR);
        }
        this.streamOptions = options;
        return this.cursor.stream(options);
    }
    /** @internal */
    _setIsEmitter() {
        if (this[kMode] === 'iterator') {
            // TODO(NODE-3485): Replace with MongoChangeStreamModeError
            throw new error_1.MongoAPIError('ChangeStream cannot be used as an EventEmitter after being used as an iterator');
        }
        this[kMode] = 'emitter';
    }
    /** @internal */
    _setIsIterator() {
        if (this[kMode] === 'emitter') {
            // TODO(NODE-3485): Replace with MongoChangeStreamModeError
            throw new error_1.MongoAPIError('ChangeStream cannot be used as an iterator after being used as an EventEmitter');
        }
        this[kMode] = 'iterator';
    }
    /**
     * Create a new change stream cursor based on self's configuration
     * @internal
     */
    _createChangeStreamCursor(options) {
        const changeStreamStageOptions = (0, utils_1.filterOptions)(options, CHANGE_STREAM_OPTIONS);
        if (this.type === CHANGE_DOMAIN_TYPES.CLUSTER) {
            changeStreamStageOptions.allChangesForCluster = true;
        }
        const pipeline = [{ $changeStream: changeStreamStageOptions }, ...this.pipeline];
        const client = this.type === CHANGE_DOMAIN_TYPES.CLUSTER
            ? this.parent
            : this.type === CHANGE_DOMAIN_TYPES.DATABASE
                ? this.parent.s.client
                : this.type === CHANGE_DOMAIN_TYPES.COLLECTION
                    ? this.parent.s.db.s.client
                    : null;
        if (client == null) {
            // This should never happen because of the assertion in the constructor
            throw new error_1.MongoRuntimeError(`Changestream type should only be one of cluster, database, collection. Found ${this.type.toString()}`);
        }
        const changeStreamCursor = new change_stream_cursor_1.ChangeStreamCursor(client, this.namespace, pipeline, options);
        for (const event of CHANGE_STREAM_EVENTS) {
            changeStreamCursor.on(event, e => this.emit(event, e));
        }
        if (this.listenerCount(ChangeStream.CHANGE) > 0) {
            this._streamEvents(changeStreamCursor);
        }
        return changeStreamCursor;
    }
    /** @internal */
    _closeEmitterModeWithError(error) {
        this.emit(ChangeStream.ERROR, error);
        this.close(() => {
            // nothing to do
        });
    }
    /** @internal */
    _streamEvents(cursor) {
        var _a;
        this._setIsEmitter();
        const stream = (_a = this[kCursorStream]) !== null && _a !== void 0 ? _a : cursor.stream();
        this[kCursorStream] = stream;
        stream.on('data', change => {
            try {
                const processedChange = this._processChange(change);
                this.emit(ChangeStream.CHANGE, processedChange);
            }
            catch (error) {
                this.emit(ChangeStream.ERROR, error);
            }
        });
        stream.on('error', error => this._processErrorStreamMode(error));
    }
    /** @internal */
    _endStream() {
        const cursorStream = this[kCursorStream];
        if (cursorStream) {
            ['data', 'close', 'end', 'error'].forEach(event => cursorStream.removeAllListeners(event));
            cursorStream.destroy();
        }
        this[kCursorStream] = undefined;
    }
    /** @internal */
    _processChange(change) {
        if (this[kClosed]) {
            // TODO(NODE-3485): Replace with MongoChangeStreamClosedError
            throw new error_1.MongoAPIError(CHANGESTREAM_CLOSED_ERROR);
        }
        // a null change means the cursor has been notified, implicitly closing the change stream
        if (change == null) {
            // TODO(NODE-3485): Replace with MongoChangeStreamClosedError
            throw new error_1.MongoRuntimeError(CHANGESTREAM_CLOSED_ERROR);
        }
        if (change && !change._id) {
            throw new error_1.MongoChangeStreamError(NO_RESUME_TOKEN_ERROR);
        }
        // cache the resume token
        this.cursor.cacheResumeToken(change._id);
        // wipe the startAtOperationTime if there was one so that there won't be a conflict
        // between resumeToken and startAtOperationTime if we need to reconnect the cursor
        this.options.startAtOperationTime = undefined;
        return change;
    }
    /** @internal */
    _processErrorStreamMode(changeStreamError) {
        // If the change stream has been closed explicitly, do not process error.
        if (this[kClosed])
            return;
        if ((0, error_1.isResumableError)(changeStreamError, this.cursor.maxWireVersion)) {
            this._endStream();
            this.cursor.close().catch(() => null);
            const topology = (0, utils_1.getTopology)(this.parent);
            topology.selectServer(this.cursor.readPreference, {}, serverSelectionError => {
                if (serverSelectionError)
                    return this._closeEmitterModeWithError(changeStreamError);
                this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
            });
        }
        else {
            this._closeEmitterModeWithError(changeStreamError);
        }
    }
    /** @internal */
    _processErrorIteratorModeCallback(changeStreamError, callback) {
        if (this[kClosed]) {
            // TODO(NODE-3485): Replace with MongoChangeStreamClosedError
            return callback(new error_1.MongoAPIError(CHANGESTREAM_CLOSED_ERROR));
        }
        if ((0, error_1.isResumableError)(changeStreamError, this.cursor.maxWireVersion)) {
            this.cursor.close().catch(() => null);
            const topology = (0, utils_1.getTopology)(this.parent);
            topology.selectServer(this.cursor.readPreference, {}, serverSelectionError => {
                // if the topology can't reconnect, close the stream
                if (serverSelectionError)
                    return this.close(() => callback(changeStreamError));
                this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
                callback();
            });
        }
        else {
            this.close(() => callback(changeStreamError));
        }
    }
}
exports.ChangeStream = ChangeStream;
/** @event */
ChangeStream.RESPONSE = constants_1.RESPONSE;
/** @event */
ChangeStream.MORE = constants_1.MORE;
/** @event */
ChangeStream.INIT = constants_1.INIT;
/** @event */
ChangeStream.CLOSE = constants_1.CLOSE;
/**
 * Fired for each new matching change in the specified namespace. Attaching a `change`
 * event listener to a Change Stream will switch the stream into flowing mode. Data will
 * then be passed as soon as it is available.
 * @event
 */
ChangeStream.CHANGE = constants_1.CHANGE;
/** @event */
ChangeStream.END = constants_1.END;
/** @event */
ChangeStream.ERROR = constants_1.ERROR;
/**
 * Emitted each time the change stream stores a new resume token.
 * @event
 */
ChangeStream.RESUME_TOKEN_CHANGED = constants_1.RESUME_TOKEN_CHANGED;
//# sourceMappingURL=change_stream.js.map