<commit_before>/*@flow*/
/* jshint esversion: 6 */
/* globals Buffer */
var Fs = require("fs");
var Fse = require("fs-extra");
var Path = require("path");
var nThen = require("nthen");
var Semaphore = require("saferphore");
var Util = require("../common-util");
var Meta = require("../metadata");
var Extras = require("../hk-util");

const readFileBin = require("../stream-file").readFileBin;
const BatchRead = require("../batch-read");

const Schedule = require("../schedule");

/*  Each time you write to a channel it will either use an open file descriptor
    for that channel or open a new descriptor if one is not available. These are
    automatically closed after this window to prevent a file descriptor leak, so
    writes that take longer than this time may be dropped! */
const CHANNEL_WRITE_WINDOW = 300000;

/*  Each time you read a channel it will have this many milliseconds to complete
    otherwise it will be closed to prevent a file descriptor leak. The server will
    lock up if it uses all available file descriptors, so it's important to close
    them. The tradeoff with this timeout is that some functions, the stream, and
    and the timeout itself are stored in memory. A longer timeout uses more memory
    and running out of memory will also kill the server. */
const STREAM_CLOSE_TIMEOUT = 120000;

/*  The above timeout closes the stream, but apparently that doesn't always work.
    We set yet another timeout to allow the runtime to gracefully close the stream
    (flushing all pending writes/reads and doing who knows what else). After this timeout
    it will be MERCILESSLY DESTROYED. This isn't graceful, but again, file descriptor
    leaks are bad. */
const STREAM_DESTROY_TIMEOUT = 30000;

const isValidChannelId = function (id) {
    return typeof(id) === 'string' &&
        id.length >= 32 && id.length < 50 &&
        /^[a-zA-Z0-9=+-]*$/.test(id);
};

// 511 -> octal 777
// read, write, execute permissions flag
const PERMISSIVE = 511;

var mkPath = function (env, channelId) {
    return Path.join(env.root, channelId.slice(0, 2), channelId) + '.ndjson';
};

var mkArchivePath = function (env, channelId) {
    return Path.join(env.archiveRoot, 'datastore', channelId.slice(0, 2), channelId) + '.ndjson';
};

var mkMetadataPath = function (env, channelId) {
    return Path.join(env.root, channelId.slice(0, 2), channelId) + '.metadata.ndjson';
};

var mkArchiveMetadataPath = function (env, channelId) {
    return Path.join(env.archiveRoot, 'datastore', channelId.slice(0, 2), channelId) + '.metadata.ndjson';
};

var mkTempPath = function (env, channelId) {
    return mkPath(env, channelId) + '.temp';
};

var mkOffsetPath = function (env, channelId) {
    return mkPath(env, channelId) + '.offset';
};

// pass in the path so we can reuse the same function for archived files
var channelExists = function (filepath, cb) {
    Fs.stat(filepath, function (err, stat) {
        if (err) {
            if (err.code === 'ENOENT') {
                // no, the file doesn't exist
                return void cb(void 0, false);
            }
            return void cb(err);
        }
        if (!stat.isFile()) { return void cb("E_NOT_FILE"); }
        return void cb(void 0, true);
    });
};

const destroyStream = function (stream) {
    if (!stream) { return; }
    try {
        stream.close();
        if (stream.closed && stream.fd === null) { return; }
    } catch (err) {
        console.error(err);
    }
    setTimeout(function () {
        try { stream.destroy(); } catch (err) { console.error(err); }
    }, STREAM_DESTROY_TIMEOUT);
};

/*  createIdleStreamCollector

Takes a stream and returns a function to asynchronously close that stream.
Successive calls to the function will be ignored.

If the function is not called for a period of STREAM_CLOSE_TIMEOUT it will
be called automatically unless its `keepAlive` method has been invoked
in the meantime. Used to prevent file descriptor leaks in the case of
abandoned streams while closing streams which are being read very very
slowly.

XXX inform the stream consumer when it has been closed prematurely
by calling back with a TIMEOUT error or something

*/
const createIdleStreamCollector = function (stream) {
    // create a function to close the stream which takes no arguments
    // and will do nothing after being called the first time
    var collector = Util.once(Util.mkAsync(Util.bake(destroyStream, [stream])));

    // create a second function which will execute the first function after a delay
    // calling this function will reset the delay and thus keep the stream 'alive'
    collector.keepAlive = Util.throttle(collector, STREAM_CLOSE_TIMEOUT);
    collector.keepAlive();
    return collector;
};

// readMessagesBin asynchronously iterates over the messages in a channel log
// the handler for each message must call back to read more, which should mean
// that this function has a lower memory profile than our classic method
// of reading logs line by line.
// it also allows the handler to abort reading at any time
const readMessagesBin = (env, id, start, msgHandler, cb) => {
    const stream = Fs.createReadStream(mkPath(env, id), { start: start });
    const collector = createIdleStreamCollector(stream);
    const handleMessageAndKeepStreamAlive = Util.both(msgHandler, collector.keepAlive);
    const done = Util.both(cb, collector);
    return void readFileBin(stream, handleMessageAndKeepStreamAlive, done, {
        offset: start,
    });
};

// reads classic metadata from a channel log and aborts
// returns undefined if the first message was not an object (not an array)
var getMetadataAtPath = function (Env, path, _cb) {
    const stream = Fs.createReadStream(path, { start: 0 });
    const collector = createIdleStreamCollector(stream);
    var cb = Util.once(Util.mkAsync(Util.both(_cb, collector)));
    var i = 0;

    return readFileBin(stream, function (msgObj, readMore, abort) {
        collector.keepAlive();
        const line = msgObj.buff.toString('utf8');

        if (!line) {
            return readMore();
        }

        // metadata should always be on the first line or not exist in the channel at all
        if (i++ > 0) {
            //console.log("aborting");
            abort();
            return void cb();
        }
        var metadata;
        try {
            metadata = JSON.parse(line);
            // if it parses, is a truthy object, and is not an array
            // then it's what you were looking for
            if (metadata && typeof(metadata) === 'object' && !Array.isArray(metadata)) {
                return void cb(void 0, metadata);
            } else { // it parsed, but isn't metadata
                return void cb(); // call back without an error or metadata
            }
        } catch (err) {
            // if you can't parse, that's bad
            return void cb("INVALID_METADATA");
        }
        readMore();
    }, function (err) {
        cb(err);
    });
};

var closeChannel = function (env, channelName, cb) {
    if (!env.channels[channelName]) { return void cb(); }
    try {
        if (typeof(Util.find(env, [ 'channels', channelName, 'writeStream', 'close'])) === 'function') {
            var stream = env.channels[channelName].writeStream;
            destroyStream(stream, channelName);
        }
        delete env.channels[channelName];
        cb();
    } catch (err) {
        cb(err);
    }
};

var clearOffset = function (env, channelId, cb) {
    var path = mkOffsetPath(env, channelId);
    // we should always be able to recover from invalid offsets, so failure to delete them
    // is not catastrophic. Anything calling this function can optionally ignore errors it might report
    Fs.unlink(path, cb);
};

var writeOffset = function (env, channelId, data, cb) {
    var path = mkOffsetPath(env, channelId);
    var s_data;
    try {
        s_data = JSON.stringify(data);
    } catch (err) {
        return void cb(err);
    }
    Fs.writeFile(path, s_data, cb);
};

var getOffset = function (env, channelId, cb) {
    var path = mkOffsetPath(env, channelId);
    Fs.readFile(path, function (err, content) {
        if (err) { return void cb(err); }
        try {
            var json = JSON.parse(content);
            cb(void 0, json);
        } catch (err2) {
            cb(err2);
        }
    });
};

// truncates a file to the end of its metadata line
// TODO write the metadata in a dedicated file
var clearChannel = function (env, channelId, _cb) {
    var cb = Util.once(Util.mkAsync(_cb));

    var path = mkPath(env, channelId);
    getMetadataAtPath(env, path, function (e, metadata) {
        if (e) { return cb(new Error(e)); }
        if (!metadata) { return void Fs.truncate(path, 0, cb); }

        var len = JSON.stringify(metadata).length + 1;

        // as long as closeChannel is synchronous, this should not cause
        // any race conditions. truncate ought to return faster than a channel
        // can be opened and read by another user. if that turns out not to be
        // the case, we'll need to implement locking.
        closeChannel(env, channelId, function (err) {
            if (err) { cb(err); }
            Fs.truncate(path, len, function (err) {
                if (err) { return cb(err); }
                cb();
            });
        });
        clearOffset(env, channelId, function () {});
    });
};

/*  readMessages is our classic method of reading messages from the disk
    notably doesn't provide a means of aborting if you finish early.
    Internally it uses readFileBin: to avoid duplicating code and to use less memory
*/
var readMessages = function (path, msgHandler, _cb) {
    var stream = Fs.createReadStream(path, { start: 0});
    var collector = createIdleStreamCollector(stream);
    var cb = Util.once(Util.mkAsync(Util.both(_cb, collector)));

    return readFileBin(stream, function (msgObj, readMore) {
        collector.keepAlive();
        msgHandler(msgObj.buff.toString('utf8'));
        readMore();
    }, function (err) {
        cb(err);
    });
};

/*  getChannelMetadata
    reads only the metadata embedded in the first line of a channel log.
    does not necessarily provide the most up to date metadata, as it
    could have been amended
*/
var getChannelMetadata = function (Env, channelId, cb) {
    var path = mkPath(Env, channelId);

    // gets metadata embedded in a file
    getMetadataAtPath(Env, path, cb);
};

// low level method for getting just the dedicated metadata channel
var getDedicatedMetadata = function (env, channelId, handler, _cb) {
    var metadataPath = mkMetadataPath(env, channelId);
    var stream = Fs.createReadStream(metadataPath, {start: 0});

    const collector = createIdleStreamCollector(stream);
    var cb = Util.both(_cb, collector);

    readFileBin(stream, function (msgObj, readMore) {
        collector.keepAlive();
        var line = msgObj.buff.toString('utf8');
        try {
            var parsed = JSON.parse(line);
            handler(null, parsed);
        } catch (err) {
            handler(err, line);
        }
        readMore();
    }, function (err) {
        // ENOENT => there is no metadata log
        if (!err || err.code === 'ENOENT') { return void cb(); }
        // otherwise stream errors?
        cb(err);
    });
};

/*  readMetadata
    fetches the classic format of the metadata from the channel log
    if it is present, otherwise load the log of metadata amendments.
    Requires a handler to process successive lines.
*/
var readMetadata = function (env, channelId, handler, cb) {
/*

Possibilities

    1. there is no metadata because it's an old channel
    2. there is metadata in the first line of the channel, but nowhere else
    3. there is metadata in the first line of the channel as well as in a dedicated log
    4. there is no metadata in the first line of the channel. Everything is in the dedicated log

How to proceed

    1. load the first line of the channel and treat it as a metadata message if applicable
    2. load the dedicated log and treat it as an update

*/

    nThen(function (w) {
        // returns the first line of a channel, parsed...
        getChannelMetadata(env, channelId, w(function (err, data) {
            if (err) {
                // 'INVALID_METADATA' if it can't parse
                // stream errors if anything goes wrong at a lower level
                    // ENOENT (no channel here)
                return void handler(err, data);
            }
            // disregard anything that isn't a map
            if (!data || typeof(data) !== 'object' || Array.isArray(data)) { return; }

            // otherwise it's good.
            handler(null, data);
        }));
    }).nThen(function () {
        getDedicatedMetadata(env, channelId, handler, function (err) {
            if (err) {
                // stream errors?
                return void cb(err);
            }
            cb();
        });
    });
};

//  writeMetadata appends to the dedicated log of metadata amendments
var writeMetadata = function (env, channelId, data, cb) {
    var path = mkMetadataPath(env, channelId);

    Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) {
        if (err && err.code !== 'EEXIST') { return void cb(err); }

        // TODO see if we can make this any faster by using something other than appendFile
        Fs.appendFile(path, data + '\n', cb);
    });
};


// check if a file exists at $path
var checkPath = function (path, callback) { // callback's second arg is never used...
    Fs.stat(path, function (err) {
        if (!err) {
            callback(undefined, true);
            return;
        }
        if (err.code !== 'ENOENT') {
            callback(err);
            return;
        }
        Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) {
            if (err && err.code !== 'EEXIST') {
                callback(err);
                return;
            }
            callback(undefined, false);
        });
    });
};

var labelError = function (label, err) {
    return label + (err.code ? "_" +  err.code: '');
};

/*  removeChannel
    fully deletes a channel log and any associated metadata
*/
var removeChannel = function (env, channelName, cb) {
    var channelPath = mkPath(env, channelName);
    var metadataPath = mkMetadataPath(env, channelName);

    var CB = Util.once(cb);

    var errors = 0;
    nThen(function (w) {
        Fs.unlink(channelPath, w(function (err) {
            if (err) {
                if (err.code === 'ENOENT') {
                    errors++;
                    return;
                }
                w.abort();
                CB(labelError("E_CHANNEL_REMOVAL", err));
            }
        }));
        Fs.unlink(metadataPath, w(function (err) {
            if (err) {
                if (err.code === 'ENOENT') {
                    errors++;
                    return;
                } // proceed if there's no metadata to delete
                w.abort();
                CB(labelError("E_METADATA_REMOVAL", err));
            }
        }));
        clearOffset(env, channelName, w());
    }).nThen(function () {
        if (errors === 2) {
            return void CB(labelError('E_REMOVE_CHANNEL', new Error("ENOENT")));
        }

        CB();
    });
};

/*  removeArchivedChannel
    fully removes an archived channel log and any associated metadata
*/
var removeArchivedChannel = function (env, channelName, cb) {
    var channelPath = mkArchivePath(env, channelName);
    var metadataPath = mkArchiveMetadataPath(env, channelName);

    var CB = Util.once(cb);

    nThen(function (w) {
        Fs.unlink(channelPath, w(function (err) {
            if (err) {
                w.abort();
                CB(labelError("E_ARCHIVED_CHANNEL_REMOVAL", err));
            }
        }));
        Fs.unlink(metadataPath, w(function (err) {
            if (err) {
                if (err.code === "ENOENT") { return; }
                w.abort();
                CB(labelError("E_ARCHIVED_METADATA_REMOVAL", err));
            }
        }));
    }).nThen(function () {
        CB();
    });
};

var _getStats = function (metadataPath, filePath, channel, cb, isLonelyMetadata) {
    var metaStat, channelStat;
    var metaErr, channelErr;
    nThen(function (ww) {
        // get the stats for the metadata
        Fs.stat(metadataPath, ww(function (err, stats) {
            if (err) {
                metaErr = err;
                return;
            }
            metaStat = stats;
        }));

        if (isLonelyMetadata) { return; }

        Fs.stat(filePath, ww(function (err, stats) {
            if (err) {
                channelErr = err;
                return;
            }
            channelStat = stats;
        }));
    }).nThen(function () {
        if (channelErr && metaErr) {
            return void cb(channelErr);
        }

        var data = {
            channel: channel,
        };

        if (metaStat && channelStat) {
        // take max of times returned by either stat
            data.atime = Math.max(channelStat.atime, metaStat.atime);
            data.mtime = Math.max(channelStat.mtime, metaStat.mtime);
            data.ctime = Math.max(channelStat.ctime, metaStat.ctime);
        // return the sum of the size of the two files
            data.size = channelStat.size + metaStat.size;
        } else if (metaStat) {
            data.atime = metaStat.atime;
            data.mtime = metaStat.mtime;
            data.ctime = metaStat.ctime;
            data.size = metaStat.size;
        } else if (channelStat) {
            data.atime = channelStat.atime;
            data.mtime = channelStat.mtime;
            data.ctime = channelStat.ctime;
            data.size = channelStat.size;
        } else {
            return void cb('NO_DATA');
        }
        cb(void 0, data);
    });
};

var getStats = function (env, channelName, cb) {
    var metadataPath = mkMetadataPath(env, channelName);
    var filePath = mkPath(env, channelName);
    _getStats(metadataPath, filePath, channelName, cb);
};

// TODO use ../plan.js for a smaller memory footprint
var listChannels = function (root, handler, cb, fast) {
    // do twenty things at a time
    var sema = Semaphore.create(20);

    var dirList = [];

    nThen(function (w) {
        // the root of your datastore contains nested directories...
        Fs.readdir(root, w(function (err, list) {
            if (err) {
                w.abort();
                // TODO check if we normally return strings or errors
                return void cb(err);
            }
            dirList = list;
        }));
    }).nThen(function (w) {
        // search inside the nested directories
        // stream it so you don't put unnecessary data in memory
        var wait = w();
        dirList.forEach(function (dir) {
            sema.take(function (give) {
    // TODO modify the asynchronous bits here to keep less in memory at any given time
    // list a directory -> process its contents with semaphores until less than N jobs are running
    // then list the next directory...
                var nestedDirPath = Path.join(root, dir);
                Fs.readdir(nestedDirPath, w(give(function (err, list) {
                    if (err) { return void handler(err); } // Is this correct?

                    list.forEach(function (item) {
                        // ignore hidden files
                        if (/^\./.test(item)) { return; }
                        // ignore anything that isn't channel or metadata
                        if (!/^[0-9a-fA-F]{32}(\.metadata?)*\.ndjson$/.test(item)) { return; }

                        var isLonelyMetadata = false;
                        var channelName;
                        var metadataName;

                        // if the current file is not the channel data, then it must be metadata
                        if (!/^[0-9a-fA-F]{32}\.ndjson$/.test(item)) {
                            metadataName = item;

                            channelName = item.replace(/\.metadata/, '');

                            // if there is a corresponding channel present in the list,
                            // then we should stop here and handle everything when we get to the channel
                            if (list.indexOf(channelName) !== -1) { return; }
                            // otherwise set a flag indicating that we should
                            // handle the metadata on its own
                            isLonelyMetadata = true;
                        } else {
                            channelName = item;
                            metadataName = channelName.replace(/\.ndjson$/, '.metadata.ndjson');
                        }

                        var channel = metadataName.replace(/\.metadata.ndjson$/, '');
                        if ([32, 34, 44].indexOf(channel.length) === -1) { return; }

                        // otherwise throw it on the pile
                        sema.take(function (give) {
                            var next = w(give());

                            if (fast) {
                                return void handler(void 0, { channel: channel, }, next);
                            }

                            var filePath = Path.join(nestedDirPath, channelName);
                            var metadataPath = Path.join(nestedDirPath, metadataName);
                            return void _getStats(metadataPath, filePath, channel, function (err, data) {
                                if (err) {
                                    return void handler(err, void 0, next);
                                }
                                handler(void 0, data, next);
                            }, isLonelyMetadata);
                        });
                    });
                })));
            });
        });
        wait();
    }).nThen(function () {
        cb();
    });
};

// move a channel's log file from its current location
// to an equivalent location in the cold storage directory
var archiveChannel = function (env, channelName, cb) {
    // TODO close channels before archiving them?

    // ctime is the most reliable indicator of when a file was archived
    // because it is used to indicate changes to the files metadata
    // and not its contents
    // if we find that this is not reliable in production, we can update it manually
    // https://nodejs.org/api/fs.html#fs_fs_utimes_path_atime_mtime_callback

    // check what the channel's path should be (in its current location)
    var currentPath = mkPath(env, channelName);

    // construct a parallel path in the new location
    var archivePath = mkArchivePath(env, channelName);

    // use Fse.move to move it, Fse makes paths to the directory when you use it.
    // https://github.com/jprichardson/node-fs-extra/blob/HEAD/docs/move.md
    nThen(function (w) {
        // move the channel log and abort if anything goes wrong
        Fse.move(currentPath, archivePath, { overwrite: true }, w(function (err) {
            if (err) {
                // proceed to the next block to remove metadata even if there's no channel
                if (err.code === 'ENOENT') { return; }
                // abort and callback for other types of errors
                w.abort();
                return void cb(err);
            }
        }));
    }).nThen(function (w) {
        clearOffset(env, channelName, w());
    }).nThen(function (w) {
        // archive the dedicated metadata channel
        var metadataPath = mkMetadataPath(env, channelName);
        var archiveMetadataPath = mkArchiveMetadataPath(env, channelName);

        Fse.move(metadataPath, archiveMetadataPath, { overwrite: true, }, w(function (err) {
            // there's no metadata to archive, so you're done!
            if (err && err.code === "ENOENT") {
                return void cb();
            }

            // there was an error archiving the metadata
            if (err) {
                return void cb(labelError("E_METADATA_ARCHIVAL", err));
            }

            // it was archived successfully
            cb();
        }));
    });
};

// restore a channel and its metadata from the archive
// to the appropriate location in the live database
var unarchiveChannel = function (env, channelName, cb) {
    // very much like 'archiveChannel' but in the opposite direction

    // the file is currently archived
    var channelPath = mkPath(env, channelName);
    var metadataPath = mkMetadataPath(env, channelName);

    // don't call the callback multiple times
    var CB = Util.once(cb);

    // if a file exists in the unarchived path, you probably don't want to clobber its data
    // so unlike 'archiveChannel' we won't overwrite.
    // Fse.move will call back with EEXIST in such a situation

    nThen(function (w) {
        // if either metadata or a file exist in prod, abort
        channelExists(channelPath, w(function (err, exists) {
            if (err) {
                w.abort();
                return void CB(err);
            }
            if (exists) {
                w.abort();
                return CB('UNARCHIVE_CHANNEL_CONFLICT');
            }
        }));
        channelExists(metadataPath, w(function (err, exists) {
            if (err) {
                w.abort();
                return void CB(err);
            }
            if (exists) {
                w.abort();
                return CB("UNARCHIVE_METADATA_CONFLICT");
            }
        }));
    }).nThen(function (w) {
        // construct archive paths
        var archiveChannelPath = mkArchivePath(env, channelName);
        // restore the archived channel
        Fse.move(archiveChannelPath, channelPath, w(function (err) {
            if (err) {
                w.abort();
                return void CB(err);
            }
        }));
    }).nThen(function (w) {
        var archiveMetadataPath = mkArchiveMetadataPath(env, channelName);
        // TODO validate that it's ok to move metadata non-atomically

        // restore the metadata log
        Fse.move(archiveMetadataPath, metadataPath, w(function (err) {
            // if there's nothing to move, you're done.
            /*if (err && err.code === 'ENOENT') {
                return CB();
            }*/ // XXX make sure removing this part won't break anything
            // call back with an error if something goes wrong
            if (err) {
                w.abort();
                return void CB(labelError("E_METADATA_RESTORATION", err));
            }
            // otherwise it was moved successfully
            CB();
        }));
    });
};

/*  channelBytes
    calls back with an error or the size (in bytes) of a channel and its metadata
*/
var channelBytes = function (env, chanName, cb) {
    var channelPath = mkPath(env, chanName);<commit_msg>suppress ENOENT on metadata when restoring archived files<commit_after>/*@flow*/
/* jshint esversion: 6 */
/* globals Buffer */
var Fs = require("fs");
var Fse = require("fs-extra");
var Path = require("path");
var nThen = require("nthen");
var Semaphore = require("saferphore");
var Util = require("../common-util");
var Meta = require("../metadata");
var Extras = require("../hk-util");

const readFileBin = require("../stream-file").readFileBin;
const BatchRead = require("../batch-read");

const Schedule = require("../schedule");

/*  Each time you write to a channel it will either use an open file descriptor
    for that channel or open a new descriptor if one is not available. These are
    automatically closed after this window to prevent a file descriptor leak, so
    writes that take longer than this time may be dropped! */
const CHANNEL_WRITE_WINDOW = 300000;

/*  Each time you read a channel it will have this many milliseconds to complete
    otherwise it will be closed to prevent a file descriptor leak. The server will
    lock up if it uses all available file descriptors, so it's important to close
    them. The tradeoff with this timeout is that some functions, the stream, and
    and the timeout itself are stored in memory. A longer timeout uses more memory
    and running out of memory will also kill the server. */
const STREAM_CLOSE_TIMEOUT = 120000;

/*  The above timeout closes the stream, but apparently that doesn't always work.
    We set yet another timeout to allow the runtime to gracefully close the stream
    (flushing all pending writes/reads and doing who knows what else). After this timeout
    it will be MERCILESSLY DESTROYED. This isn't graceful, but again, file descriptor
    leaks are bad. */
const STREAM_DESTROY_TIMEOUT = 30000;

const isValidChannelId = function (id) {
    return typeof(id) === 'string' &&
        id.length >= 32 && id.length < 50 &&
        /^[a-zA-Z0-9=+-]*$/.test(id);
};

// 511 -> octal 777
// read, write, execute permissions flag
const PERMISSIVE = 511;

var mkPath = function (env, channelId) {
    return Path.join(env.root, channelId.slice(0, 2), channelId) + '.ndjson';
};

var mkArchivePath = function (env, channelId) {
    return Path.join(env.archiveRoot, 'datastore', channelId.slice(0, 2), channelId) + '.ndjson';
};

var mkMetadataPath = function (env, channelId) {
    return Path.join(env.root, channelId.slice(0, 2), channelId) + '.metadata.ndjson';
};

var mkArchiveMetadataPath = function (env, channelId) {
    return Path.join(env.archiveRoot, 'datastore', channelId.slice(0, 2), channelId) + '.metadata.ndjson';
};

var mkTempPath = function (env, channelId) {
    return mkPath(env, channelId) + '.temp';
};

var mkOffsetPath = function (env, channelId) {
    return mkPath(env, channelId) + '.offset';
};

// pass in the path so we can reuse the same function for archived files
var channelExists = function (filepath, cb) {
    Fs.stat(filepath, function (err, stat) {
        if (err) {
            if (err.code === 'ENOENT') {
                // no, the file doesn't exist
                return void cb(void 0, false);
            }
            return void cb(err);
        }
        if (!stat.isFile()) { return void cb("E_NOT_FILE"); }
        return void cb(void 0, true);
    });
};

const destroyStream = function (stream) {
    if (!stream) { return; }
    try {
        stream.close();
        if (stream.closed && stream.fd === null) { return; }
    } catch (err) {
        console.error(err);
    }
    setTimeout(function () {
        try { stream.destroy(); } catch (err) { console.error(err); }
    }, STREAM_DESTROY_TIMEOUT);
};

/*  createIdleStreamCollector

Takes a stream and returns a function to asynchronously close that stream.
Successive calls to the function will be ignored.

If the function is not called for a period of STREAM_CLOSE_TIMEOUT it will
be called automatically unless its `keepAlive` method has been invoked
in the meantime. Used to prevent file descriptor leaks in the case of
abandoned streams while closing streams which are being read very very
slowly.

XXX inform the stream consumer when it has been closed prematurely
by calling back with a TIMEOUT error or something

*/
const createIdleStreamCollector = function (stream) {
    // create a function to close the stream which takes no arguments
    // and will do nothing after being called the first time
    var collector = Util.once(Util.mkAsync(Util.bake(destroyStream, [stream])));

    // create a second function which will execute the first function after a delay
    // calling this function will reset the delay and thus keep the stream 'alive'
    collector.keepAlive = Util.throttle(collector, STREAM_CLOSE_TIMEOUT);
    collector.keepAlive();
    return collector;
};

// readMessagesBin asynchronously iterates over the messages in a channel log
// the handler for each message must call back to read more, which should mean
// that this function has a lower memory profile than our classic method
// of reading logs line by line.
// it also allows the handler to abort reading at any time
const readMessagesBin = (env, id, start, msgHandler, cb) => {
    const stream = Fs.createReadStream(mkPath(env, id), { start: start });
    const collector = createIdleStreamCollector(stream);
    const handleMessageAndKeepStreamAlive = Util.both(msgHandler, collector.keepAlive);
    const done = Util.both(cb, collector);
    return void readFileBin(stream, handleMessageAndKeepStreamAlive, done, {
        offset: start,
    });
};

// reads classic metadata from a channel log and aborts
// returns undefined if the first message was not an object (not an array)
var getMetadataAtPath = function (Env, path, _cb) {
    const stream = Fs.createReadStream(path, { start: 0 });
    const collector = createIdleStreamCollector(stream);
    var cb = Util.once(Util.mkAsync(Util.both(_cb, collector)));
    var i = 0;

    return readFileBin(stream, function (msgObj, readMore, abort) {
        collector.keepAlive();
        const line = msgObj.buff.toString('utf8');

        if (!line) {
            return readMore();
        }

        // metadata should always be on the first line or not exist in the channel at all
        if (i++ > 0) {
            //console.log("aborting");
            abort();
            return void cb();
        }
        var metadata;
        try {
            metadata = JSON.parse(line);
            // if it parses, is a truthy object, and is not an array
            // then it's what you were looking for
            if (metadata && typeof(metadata) === 'object' && !Array.isArray(metadata)) {
                return void cb(void 0, metadata);
            } else { // it parsed, but isn't metadata
                return void cb(); // call back without an error or metadata
            }
        } catch (err) {
            // if you can't parse, that's bad
            return void cb("INVALID_METADATA");
        }
        readMore();
    }, function (err) {
        cb(err);
    });
};

var closeChannel = function (env, channelName, cb) {
    if (!env.channels[channelName]) { return void cb(); }
    try {
        if (typeof(Util.find(env, [ 'channels', channelName, 'writeStream', 'close'])) === 'function') {
            var stream = env.channels[channelName].writeStream;
            destroyStream(stream, channelName);
        }
        delete env.channels[channelName];
        cb();
    } catch (err) {
        cb(err);
    }
};

var clearOffset = function (env, channelId, cb) {
    var path = mkOffsetPath(env, channelId);
    // we should always be able to recover from invalid offsets, so failure to delete them
    // is not catastrophic. Anything calling this function can optionally ignore errors it might report
    Fs.unlink(path, cb);
};

var writeOffset = function (env, channelId, data, cb) {
    var path = mkOffsetPath(env, channelId);
    var s_data;
    try {
        s_data = JSON.stringify(data);
    } catch (err) {
        return void cb(err);
    }
    Fs.writeFile(path, s_data, cb);
};

var getOffset = function (env, channelId, cb) {
    var path = mkOffsetPath(env, channelId);
    Fs.readFile(path, function (err, content) {
        if (err) { return void cb(err); }
        try {
            var json = JSON.parse(content);
            cb(void 0, json);
        } catch (err2) {
            cb(err2);
        }
    });
};

// truncates a file to the end of its metadata line
// TODO write the metadata in a dedicated file
var clearChannel = function (env, channelId, _cb) {
    var cb = Util.once(Util.mkAsync(_cb));

    var path = mkPath(env, channelId);
    getMetadataAtPath(env, path, function (e, metadata) {
        if (e) { return cb(new Error(e)); }
        if (!metadata) { return void Fs.truncate(path, 0, cb); }

        var len = JSON.stringify(metadata).length + 1;

        // as long as closeChannel is synchronous, this should not cause
        // any race conditions. truncate ought to return faster than a channel
        // can be opened and read by another user. if that turns out not to be
        // the case, we'll need to implement locking.
        closeChannel(env, channelId, function (err) {
            if (err) { cb(err); }
            Fs.truncate(path, len, function (err) {
                if (err) { return cb(err); }
                cb();
            });
        });
        clearOffset(env, channelId, function () {});
    });
};

/*  readMessages is our classic method of reading messages from the disk
    notably doesn't provide a means of aborting if you finish early.
    Internally it uses readFileBin: to avoid duplicating code and to use less memory
*/
var readMessages = function (path, msgHandler, _cb) {
    var stream = Fs.createReadStream(path, { start: 0});
    var collector = createIdleStreamCollector(stream);
    var cb = Util.once(Util.mkAsync(Util.both(_cb, collector)));

    return readFileBin(stream, function (msgObj, readMore) {
        collector.keepAlive();
        msgHandler(msgObj.buff.toString('utf8'));
        readMore();
    }, function (err) {
        cb(err);
    });
};

/*  getChannelMetadata
    reads only the metadata embedded in the first line of a channel log.
    does not necessarily provide the most up to date metadata, as it
    could have been amended
*/
var getChannelMetadata = function (Env, channelId, cb) {
    var path = mkPath(Env, channelId);

    // gets metadata embedded in a file
    getMetadataAtPath(Env, path, cb);
};

// low level method for getting just the dedicated metadata channel
var getDedicatedMetadata = function (env, channelId, handler, _cb) {
    var metadataPath = mkMetadataPath(env, channelId);
    var stream = Fs.createReadStream(metadataPath, {start: 0});

    const collector = createIdleStreamCollector(stream);
    var cb = Util.both(_cb, collector);

    readFileBin(stream, function (msgObj, readMore) {
        collector.keepAlive();
        var line = msgObj.buff.toString('utf8');
        try {
            var parsed = JSON.parse(line);
            handler(null, parsed);
        } catch (err) {
            handler(err, line);
        }
        readMore();
    }, function (err) {
        // ENOENT => there is no metadata log
        if (!err || err.code === 'ENOENT') { return void cb(); }
        // otherwise stream errors?
        cb(err);
    });
};

/*  readMetadata
    fetches the classic format of the metadata from the channel log
    if it is present, otherwise load the log of metadata amendments.
    Requires a handler to process successive lines.
*/
var readMetadata = function (env, channelId, handler, cb) {
/*

Possibilities

    1. there is no metadata because it's an old channel
    2. there is metadata in the first line of the channel, but nowhere else
    3. there is metadata in the first line of the channel as well as in a dedicated log
    4. there is no metadata in the first line of the channel. Everything is in the dedicated log

How to proceed

    1. load the first line of the channel and treat it as a metadata message if applicable
    2. load the dedicated log and treat it as an update

*/

    nThen(function (w) {
        // returns the first line of a channel, parsed...
        getChannelMetadata(env, channelId, w(function (err, data) {
            if (err) {
                // 'INVALID_METADATA' if it can't parse
                // stream errors if anything goes wrong at a lower level
                    // ENOENT (no channel here)
                return void handler(err, data);
            }
            // disregard anything that isn't a map
            if (!data || typeof(data) !== 'object' || Array.isArray(data)) { return; }

            // otherwise it's good.
            handler(null, data);
        }));
    }).nThen(function () {
        getDedicatedMetadata(env, channelId, handler, function (err) {
            if (err) {
                // stream errors?
                return void cb(err);
            }
            cb();
        });
    });
};

//  writeMetadata appends to the dedicated log of metadata amendments
var writeMetadata = function (env, channelId, data, cb) {
    var path = mkMetadataPath(env, channelId);

    Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) {
        if (err && err.code !== 'EEXIST') { return void cb(err); }

        // TODO see if we can make this any faster by using something other than appendFile
        Fs.appendFile(path, data + '\n', cb);
    });
};


// check if a file exists at $path
var checkPath = function (path, callback) { // callback's second arg is never used...
    Fs.stat(path, function (err) {
        if (!err) {
            callback(undefined, true);
            return;
        }
        if (err.code !== 'ENOENT') {
            callback(err);
            return;
        }
        Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) {
            if (err && err.code !== 'EEXIST') {
                callback(err);
                return;
            }
            callback(undefined, false);
        });
    });
};

var labelError = function (label, err) {
    return label + (err.code ? "_" +  err.code: '');
};

/*  removeChannel
    fully deletes a channel log and any associated metadata
*/
var removeChannel = function (env, channelName, cb) {
    var channelPath = mkPath(env, channelName);
    var metadataPath = mkMetadataPath(env, channelName);

    var CB = Util.once(cb);

    var errors = 0;
    nThen(function (w) {
        Fs.unlink(channelPath, w(function (err) {
            if (err) {
                if (err.code === 'ENOENT') {
                    errors++;
                    return;
                }
                w.abort();
                CB(labelError("E_CHANNEL_REMOVAL", err));
            }
        }));
        Fs.unlink(metadataPath, w(function (err) {
            if (err) {
                if (err.code === 'ENOENT') {
                    errors++;
                    return;
                } // proceed if there's no metadata to delete
                w.abort();
                CB(labelError("E_METADATA_REMOVAL", err));
            }
        }));
        clearOffset(env, channelName, w());
    }).nThen(function () {
        if (errors === 2) {
            return void CB(labelError('E_REMOVE_CHANNEL', new Error("ENOENT")));
        }

        CB();
    });
};

/*  removeArchivedChannel
    fully removes an archived channel log and any associated metadata
*/
var removeArchivedChannel = function (env, channelName, cb) {
    var channelPath = mkArchivePath(env, channelName);
    var metadataPath = mkArchiveMetadataPath(env, channelName);

    var CB = Util.once(cb);

    nThen(function (w) {
        Fs.unlink(channelPath, w(function (err) {
            if (err) {
                w.abort();
                CB(labelError("E_ARCHIVED_CHANNEL_REMOVAL", err));
            }
        }));
        Fs.unlink(metadataPath, w(function (err) {
            if (err) {
                if (err.code === "ENOENT") { return; }
                w.abort();
                CB(labelError("E_ARCHIVED_METADATA_REMOVAL", err));
            }
        }));
    }).nThen(function () {
        CB();
    });
};

var _getStats = function (metadataPath, filePath, channel, cb, isLonelyMetadata) {
    var metaStat, channelStat;
    var metaErr, channelErr;
    nThen(function (ww) {
        // get the stats for the metadata
        Fs.stat(metadataPath, ww(function (err, stats) {
            if (err) {
                metaErr = err;
                return;
            }
            metaStat = stats;
        }));

        if (isLonelyMetadata) { return; }

        Fs.stat(filePath, ww(function (err, stats) {
            if (err) {
                channelErr = err;
                return;
            }
            channelStat = stats;
        }));
    }).nThen(function () {
        if (channelErr && metaErr) {
            return void cb(channelErr);
        }

        var data = {
            channel: channel,
        };

        if (metaStat && channelStat) {
        // take max of times returned by either stat
            data.atime = Math.max(channelStat.atime, metaStat.atime);
            data.mtime = Math.max(channelStat.mtime, metaStat.mtime);
            data.ctime = Math.max(channelStat.ctime, metaStat.ctime);
        // return the sum of the size of the two files
            data.size = channelStat.size + metaStat.size;
        } else if (metaStat) {
            data.atime = metaStat.atime;
            data.mtime = metaStat.mtime;
            data.ctime = metaStat.ctime;
            data.size = metaStat.size;
        } else if (channelStat) {
            data.atime = channelStat.atime;
            data.mtime = channelStat.mtime;
            data.ctime = channelStat.ctime;
            data.size = channelStat.size;
        } else {
            return void cb('NO_DATA');
        }
        cb(void 0, data);
    });
};

var getStats = function (env, channelName, cb) {
    var metadataPath = mkMetadataPath(env, channelName);
    var filePath = mkPath(env, channelName);
    _getStats(metadataPath, filePath, channelName, cb);
};

// TODO use ../plan.js for a smaller memory footprint
var listChannels = function (root, handler, cb, fast) {
    // do twenty things at a time
    var sema = Semaphore.create(20);

    var dirList = [];

    nThen(function (w) {
        // the root of your datastore contains nested directories...
        Fs.readdir(root, w(function (err, list) {
            if (err) {
                w.abort();
                // TODO check if we normally return strings or errors
                return void cb(err);
            }
            dirList = list;
        }));
    }).nThen(function (w) {
        // search inside the nested directories
        // stream it so you don't put unnecessary data in memory
        var wait = w();
        dirList.forEach(function (dir) {
            sema.take(function (give) {
    // TODO modify the asynchronous bits here to keep less in memory at any given time
    // list a directory -> process its contents with semaphores until less than N jobs are running
    // then list the next directory...
                var nestedDirPath = Path.join(root, dir);
                Fs.readdir(nestedDirPath, w(give(function (err, list) {
                    if (err) { return void handler(err); } // Is this correct?

                    list.forEach(function (item) {
                        // ignore hidden files
                        if (/^\./.test(item)) { return; }
                        // ignore anything that isn't channel or metadata
                        if (!/^[0-9a-fA-F]{32}(\.metadata?)*\.ndjson$/.test(item)) { return; }

                        var isLonelyMetadata = false;
                        var channelName;
                        var metadataName;

                        // if the current file is not the channel data, then it must be metadata
                        if (!/^[0-9a-fA-F]{32}\.ndjson$/.test(item)) {
                            metadataName = item;

                            channelName = item.replace(/\.metadata/, '');

                            // if there is a corresponding channel present in the list,
                            // then we should stop here and handle everything when we get to the channel
                            if (list.indexOf(channelName) !== -1) { return; }
                            // otherwise set a flag indicating that we should
                            // handle the metadata on its own
                            isLonelyMetadata = true;
                        } else {
                            channelName = item;
                            metadataName = channelName.replace(/\.ndjson$/, '.metadata.ndjson');
                        }

                        var channel = metadataName.replace(/\.metadata.ndjson$/, '');
                        if ([32, 34, 44].indexOf(channel.length) === -1) { return; }

                        // otherwise throw it on the pile
                        sema.take(function (give) {
                            var next = w(give());

                            if (fast) {
                                return void handler(void 0, { channel: channel, }, next);
                            }

                            var filePath = Path.join(nestedDirPath, channelName);
                            var metadataPath = Path.join(nestedDirPath, metadataName);
                            return void _getStats(metadataPath, filePath, channel, function (err, data) {
                                if (err) {
                                    return void handler(err, void 0, next);
                                }
                                handler(void 0, data, next);
                            }, isLonelyMetadata);
                        });
                    });
                })));
            });
        });
        wait();
    }).nThen(function () {
        cb();
    });
};

// move a channel's log file from its current location
// to an equivalent location in the cold storage directory
var archiveChannel = function (env, channelName, cb) {
    // TODO close channels before archiving them?

    // ctime is the most reliable indicator of when a file was archived
    // because it is used to indicate changes to the files metadata
    // and not its contents
    // if we find that this is not reliable in production, we can update it manually
    // https://nodejs.org/api/fs.html#fs_fs_utimes_path_atime_mtime_callback

    // check what the channel's path should be (in its current location)
    var currentPath = mkPath(env, channelName);

    // construct a parallel path in the new location
    var archivePath = mkArchivePath(env, channelName);

    // use Fse.move to move it, Fse makes paths to the directory when you use it.
    // https://github.com/jprichardson/node-fs-extra/blob/HEAD/docs/move.md
    nThen(function (w) {
        // move the channel log and abort if anything goes wrong
        Fse.move(currentPath, archivePath, { overwrite: true }, w(function (err) {
            if (err) {
                // proceed to the next block to remove metadata even if there's no channel
                if (err.code === 'ENOENT') { return; }
                // abort and callback for other types of errors
                w.abort();
                return void cb(err);
            }
        }));
    }).nThen(function (w) {
        clearOffset(env, channelName, w());
    }).nThen(function (w) {
        // archive the dedicated metadata channel
        var metadataPath = mkMetadataPath(env, channelName);
        var archiveMetadataPath = mkArchiveMetadataPath(env, channelName);

        Fse.move(metadataPath, archiveMetadataPath, { overwrite: true, }, w(function (err) {
            // there's no metadata to archive, so you're done!
            if (err && err.code === "ENOENT") {
                return void cb();
            }

            // there was an error archiving the metadata
            if (err) {
                return void cb(labelError("E_METADATA_ARCHIVAL", err));
            }

            // it was archived successfully
            cb();
        }));
    });
};

// restore a channel and its metadata from the archive
// to the appropriate location in the live database
var unarchiveChannel = function (env, channelName, cb) {
    // very much like 'archiveChannel' but in the opposite direction

    // the file is currently archived
    var channelPath = mkPath(env, channelName);
    var metadataPath = mkMetadataPath(env, channelName);

    // don't call the callback multiple times
    var CB = Util.once(cb);

    // if a file exists in the unarchived path, you probably don't want to clobber its data
    // so unlike 'archiveChannel' we won't overwrite.
    // Fse.move will call back with EEXIST in such a situation

    nThen(function (w) {
        // if either metadata or a file exist in prod, abort
        channelExists(channelPath, w(function (err, exists) {
            if (err) {
                w.abort();
                return void CB(err);
            }
            if (exists) {
                w.abort();
                return CB('UNARCHIVE_CHANNEL_CONFLICT');
            }
        }));
        channelExists(metadataPath, w(function (err, exists) {
            if (err) {
                w.abort();
                return void CB(err);
            }
            if (exists) {
                w.abort();
                return CB("UNARCHIVE_METADATA_CONFLICT");
            }
        }));
    }).nThen(function (w) {
        // construct archive paths
        var archiveChannelPath = mkArchivePath(env, channelName);
        // restore the archived channel
        Fse.move(archiveChannelPath, channelPath, w(function (err) {
            if (err) {
                w.abort();
                return void CB(err);
            }
        }));
    }).nThen(function (w) {
        var archiveMetadataPath = mkArchiveMetadataPath(env, channelName);
        // TODO validate that it's ok to move metadata non-atomically

        // restore the metadata log
        Fse.move(archiveMetadataPath, metadataPath, w(function (err) {
            // if there's nothing to move, you're done.
            if (err && err.code === 'ENOENT') {
                return CB();
            }
            // call back with an error if something goes wrong
            if (err) {
                w.abort();
                return void CB(labelError("E_METADATA_RESTORATION", err));
            }
            // otherwise it was moved successfully
            CB();
        }));
    });
};

/*  channelBytes
    calls back with an error or the size (in bytes) of a channel and its metadata
*/
var channelBytes = function (env, chanName, cb) {
    var channelPath = mkPath(env, chanName);
{'fraction_non_alphanumeric': 0.10668774997674635, 'fraction_numerical': 0.00323690819458655, 'mean_word_length': 1.9964325529542921, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5152205', 'n_tokens_mistral': 14899, 'n_tokens_neox': 14320, 'n_words': 5045}