<commit_before>'use strict'; // eslint-disable-line

const stream = require('stream');
const MongoClient = require('mongodb').MongoClient;
const { Timestamp } = require('bson');

const ops = {
    i: 'put',
    u: 'put',
    d: 'delete',
};

class ListRecordStream extends stream.Transform {
    constructor(logger, lastEndID) {
        super({ objectMode: true });
        this.logger = logger;
        this.lastEndID = lastEndID;
        this.hasStarted = false;
        this.start = null;
        this.end = null;
        this.lastUniqID = null;
        // this.unpublishedListing is true once we pass the oplog that has the
        // start seq timestamp and uniqID 'h'
        this.unpublishedListing = null;
    }

    _transform(itemObj, encoding, callback) {
        if (!itemObj) {
            this.push(null);
            this.emit('info', {
                start: this.start,
                end: this.end,
                uniqID: this.lastUniqID,
            });
            return callback();
        }

        // always update to most recent uniqID
        this.lastUniqID = itemObj.h.toString();

        if (this.end === null || itemObj.ts.toNumber() > this.end) {
            this.end = itemObj.ts.toNumber();
        }

        // only push to stream unpublished objects
        if (!this.unpublishedListing) {
            if (this.lastEndID === itemObj.h.toString()) {
                this.unpublishedListing = true;
            }
        }

        if (!this.hasStarted) {
            this.hasStarted = true;
            this.start = itemObj.ts.toNumber();
            this.emit('info', {
                start: this.start,
                end: this.end,
                uniqId: this.lastUniqID,
            });
        }

        // don't push oplogs that have already been sent
        if (!this.unpublishedListing) {
            return callback();
        }

        const dbName = itemObj.ns.split('.');
        const streamObject = {
            timestamp: new Date(itemObj.ts.high_ * 1000),
            db: dbName[1],
            entries: [
                {
                    type: ops[itemObj.op],
                    key: itemObj.o._id,
                    value: JSON.stringify(itemObj.o.value),
                },
            ],
        };
        return callback(null, streamObject);
    }

    _flush(callback) {
        this.emit('info', {
            start: this.start,
            end: this.end,
            uniqID: this.lastUniqID,
        });
        this.push(null);
        callback();
    }
}

/**
 * @class
 * @classdesc Class to consume mongo oplog<commit_msg>bf: ZENKO-344 Fix duplicate mongo logs<commit_after>'use strict'; // eslint-disable-line

const stream = require('stream');
const MongoClient = require('mongodb').MongoClient;
const { Timestamp } = require('bson');

const ops = {
    i: 'put',
    u: 'put',
    d: 'delete',
};

class ListRecordStream extends stream.Transform {
    constructor(logger, lastEndID) {
        super({ objectMode: true });
        this.logger = logger;
        this.lastEndID = lastEndID;
        this.hasStarted = false;
        this.start = null;
        this.end = null;
        this.lastUniqID = null;
        // this.unpublishedListing is true once we pass the oplog that has the
        // start seq timestamp and uniqID 'h'
        this.unpublishedListing = false;
    }

    _transform(itemObj, encoding, callback) {
        if (!itemObj) {
            this.push(null);
            this.emit('info', {
                start: this.start,
                end: this.end,
                uniqID: this.lastUniqID,
            });
            return callback();
        }

        // always update to most recent uniqID
        this.lastUniqID = itemObj.h.toString();

        if (this.end === null || itemObj.ts.toNumber() > this.end) {
            this.end = itemObj.ts.toNumber();
        }

        // only push to stream unpublished objects
        if (!this.unpublishedListing) {
            // When an oplog with a unique ID that is stored in the
            // log offset is found, all oplogs AFTER this is unpublished.
            if (this.lastEndID === itemObj.h.toString()) {
                this.unpublishedListing = true;
            }
            return callback();
        }

        if (!this.hasStarted) {
            this.hasStarted = true;
            this.start = itemObj.ts.toNumber();
            this.emit('info', {
                start: this.start,
                end: this.end,
                uniqId: this.lastUniqID,
            });
        }

        const dbName = itemObj.ns.split('.');
        const streamObject = {
            timestamp: new Date(itemObj.ts.high_ * 1000),
            db: dbName[1],
            entries: [
                {
                    type: ops[itemObj.op],
                    key: itemObj.o._id,
                    value: JSON.stringify(itemObj.o.value),
                },
            ],
        };
        return callback(null, streamObject);
    }

    _flush(callback) {
        if (!this.hasStarted) {
            this.emit('info', {
                start: this.start,
                end: this.end,
                uniqID: this.lastUniqID,
            });
        }
        this.push(null);
        callback();
    }
}

/**
 * @class
 * @classdesc Class to consume mongo oplog
{'fraction_non_alphanumeric': 0.11829771278590176, 'fraction_numerical': 0.0024371953505811774, 'mean_word_length': 1.4860205032618825, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2882642', 'n_tokens_mistral': 1559, 'n_tokens_neox': 1498, 'n_words': 350}