const leo = require('leo-sdk');
const checksum = require('leo-connector-common/checksum');
const moment = require('moment');

exports.handler = function(event, context, callback) {
    let db1 = checksum.lambdaConnector('MySQL DB Lead checksum', process.env.mysql_lambda, {
        sql: `SELECT id, status FROM orders WHERE id __IDCOLUMNLIMIT__`,
        table: 'orders',
        id_column: 'id',
        key_column: 'primary'
    });
    let db2 = checksum.lambdaConnector('Postgres DB Lead checksum', process.env.postgres_lambda, {
        sql: `SELECT id, status FROM orders WHERE id __IDCOLUMNLIMIT__`,
        table: 'orders',
        id_column: 'id',
        key_column: 'primary'
    });
    
    // checksum code in step 3 (below) goes here
}

let system = 'default';
checksum.checksum(system, event.botId, db1, db2, {
    stopOnStreak: 1750000, // Set the number of records that if the checksum finds in sequence that are identical, it will stop and mark itself as completed.
    stop_at: moment().add({minutes: 4}), // Lambda has a 5-minute limit, so we set this to 4 so the bot has time to cleanup. It will restart right after this and continue where it left off.
    limit: 20000, // the number of records to start comparing between the 2 databases.
    maxLimit: 500000, // If a "block" 20,000 or more records are identical, increase the comparison block size from limit to this max limit
    shouldDelete: false, // set this to true if you want records that exist in the slave database but not in master to be deleted.
    loadSize: 50000, // this is the recommended load size
    reverse: true, // Processes records from highest to lowest. Set to false to process from lowest to highest.
    sample: true, // 
    queue: { // this controls the queue where the ID's go that are marked as missing from the slave database
        name: event.destination, // queue name.
        transform: leo.streams.through((obj, done) => { // How to transform the ID's before sending into the queue.
            done(null, {
                Orders: obj.missing.concat(obj.incorrect)
            });
        })
    }
    //skipBatch: true, // only set to true if you need to 2 connectors to compare individual records insteadof batches
    //showOutput: false
})
.then(data=>{ console.log(data); callback()})
.catch(callback);

let customConnector = checksum.basicConnector('< Checksum name >', {
    id_column: 'id'
}, {
    // custom handlers go here
});

/**
 * Called when checksum starts.
 * Used for situations such as when your endpoint requires authorization.
 * Called with data, return a session
 */
initialize: function(data) {
    return Promise.resolve({});
}

/**
 * @int start
 * @int end
 * @object options (optional)
 * @return object {min: int, max: int, total: int}
 */
// Respond to start and end -- options
// Return object with min, max, total
range: function(start, end) {
    let min = null;
    let max = null;
    let total = 0;
    
    /************************************************
    * Begin example code to get min, max, and total.*
    * This example loops through records returned   *
    * into “db” and creates a start and end from    *
    * the greatest and least id’s.                  *
    *************************************************/
    // db: object containing records to compare
    let db = [{id: 1, name: 'foo', etc: 'etc'}, {...}];
    Object.keys(db).map(id => {
        id = db[id][this.settings.id_column];
        if ((start === undefined || id >= start) && (end === undefined || id <= end)) {
            total++;
            if (min == null || id < min) {
                min = id;
            }
            if (max == null || id > max) {
                max = id;
            }
        }
    });
    /**********************************************
    * End example code to get min, max, and total *
    ***********************************************/
    
    // return a min, max and total
    return Promise.resolve({
        min,
        max,
        total
    });
}

/**
 * Respond to a start and end, and build an array of data returned into “db”
 * 
 * @int start
 * @int end
 * @return mixed (Stream|Array|hash)
 */
batch: function(start, end) {
    let data = [];
    // db: object containing records to compare
    let db = [{id: 1, name: 'foo', etc: 'etc'}, {...}];
    
    /***********************************************************************************
     * Example code to put together an array of data using the data returned from “db” *
     ***********************************************************************************/
    for (v of db) {
        data.push(v);
    }
    
    /**********************************************************************************************************
    * Alternatively, if you cannot pass in a start and end and just get a chunk of data back, build an array *
    * with the data having id’s between start and end                                                         *
    ***********************************************************************************************************/
    for (let i = start; i <= end; i++) {
        if (typeof db[i] !== 'undefined') {
            data.push(db[i]);
        }
    }
    
    // return the array of data
    return Promise.resolve(data);
}

/**
 * Nibble handler: Uses a start, end, limit, and reverse; and gives a “next” and “current” to continue checking data
 * @int start
 * @int end
 * @int limit
 * @bool reverse
 */
// Responds to start, end, limit, reverse
// Returns object with next, current
nibble: function(start, end, limit, reverse) {
    // db: object containing records to compare
    let db = [{id: 1, name: 'foo', etc: 'etc'}, {...}];
    let current = null;
    let next = null;
    let dir = 1;
    let ostart = start;
    let oend = end;
    if (reverse) {
        start = end;
        end = ostart;
        dir = -1;
    }
    let cnt = 0;
    for (let i = start; i >= ostart && i <= oend; i += dir) {
        if (typeof db[i] !== undefined) {
            let v = db[i];
            cnt++;
            if (cnt >= limit) {
                if (!current) {
                    current = v[this.settings.id_column];
                } else {
                    next = v[this.settings.id_column];
                    break;
                }
            }
        }
    }

    return Promise.resolve({
        current,
        next
    });
}

individual: function(start, end) {
    return this.batch(start, end);
}

delete: function(ids) {
    // db: object containing records to compare
    let db = [{id: 1, name: 'foo', etc: 'etc'}, {...}];
    ids.map(id => {
        if (id in db) {
            delete db[id];
        }
    });
    return Promise.resolve();
}

// Respond to ids
// Return Stream, Array
sample: function(ids) {
    let data = [];
    // db: object containing records to compare
    let db = [{id: 1, name: 'foo', etc: 'etc'}, {...}];
    ids.map(id => {
        let v = db[id];
        if (v !== undefined) {
            data.push(v);
            console.log(v);
        }
    });

    return Promise.resolve(data);
}

destroy: function(data) {
    return Promise.resolve();
}

// include the required handlers
const leo = require("leo-sdk");
const helperFactory = require("leo-connector-common/botHelper");
// config requires leo-sdk 2.x
const config = require('leo-config');
// use the connector for your database type:
const connector = require('leo-connector-<dbtype>');

// use this handler for leo-sdk 2.x
exports.handler = require("leo-sdk/wrappers/cron.js")(async function(event, context, callback) {
    
// use this handler for leo-sdk 1.x    
exports.handler = async function(event, context, callback) {
    
    // create the helper    
    const helper = new helperFactory(event, context, leo);

"use strict";

const leo = require("leo-sdk");
const ls = require("leo-streams");
const config = require("leo-config");
const connector = require('leo-connector-mysql');

exports.handler = require("leo-sdk/wrappers/cron.js")(async function (event, context, callback) {

    let source = 'system:mysql';
    let destination = 'people';
    // these settings are specific to snapshot. 
    let snapshotSettings = {
        table: 'people', // table name that has our primary keys
        pk: 'id', // primary key of the table we're using
        limit: 5000, // max number of id's to insert into the select query
    };

    let domainObjectBuilder = connector.domainObjectBuilder(config.db);
    let snapshotter = connector.snapshotter(config.db);

    // translate id's into id's specific to the domain we're creating.
    // In the translation object below, we're going to get people.id from an array of people ids, as well as an array of contact ids.
    let tableIdTranslations = {
        people: true, // No Translation needed. Expects id's to be in format like this: {payload.[update.]people[1, 2, 3, 5, 25, 35, 49, 50, etc…]}
        contacts: data => `SELECT people_id FROM contacts WHERE id in (?)`, // Gets the people.id from the contacts table from a format like this: {payload.[update.]contacts[1, 2, 3, 5, 25, 35, 49, 50, etc…]}
    };

    let domainObject = new domainObjectBuilder.DomainObject(
        // for mysql, c.database contains the database name
        c => `SELECT
            t.id as _domain_id,
            t.*,
            'test' as myField
            FROM people AS t
            WHERE t.id IN (?)`
    );
    
    // if you have a one-to-many to join:
    domainObject.hasMany("sub_object", c => `SELECT related_table.pk as _domain_id, related_table.* from related_table where pk IN (?)`); // ? will be filled with the object primary key as laid out in the tableIdTranslation

    //Advanced Settings
    // if we're creating a snapshot
    if (process.env.snapshot) {
        ls.pipe(
            // read through primary keys
            snapshotter.read(snapshotSettings),

            // transform keys into domain objects
            dol.domainObjectTransform(domainObject),

            // load into S3, toLeo, and checkpoint
            snapshotter.write(context.botId, destination),

            // end
            err => {
                callback(err);
            }
        );
    } else {
        // regular domain object load
        ls.pipe(
            leo.read(context.botId, source),

            dol.translateIds(tableIdTranslations),
            dol.domainObjectTransform(domainObject),

            leo.load(context.botId, destination),
            (err) => {
                callback(err);
            }
        );
    }
    //End Advanced Settings
});

let query = `SELECT a.id
        , a.name
        , a.foo_id
        , '' AS prefix_Foo
        , b.id
        , b.name
    FROM tableOne a
    JOIN tableTwo b ON (b.id = a.foo_id)    
    WHERE a.id IN (?)`;