<commit_before>const filesystem = require('fs')
const fs = require('fs').promises
const events = require('events')
const Queue = require('avenue')
const path = require('path')
const assert = require('assert')
const logger = require('prolific.logger').createLogger('prolific')
const coalesce = require('extant')

const LEVEL = { ENOENT: 'debug' }

class Watcher extends events.EventEmitter {
    constructor (destructible, hash, directory) {
        super()
        this._queue = new Queue
        this._hash = hash
        this._directory = directory
        this._watcher = filesystem.watch(directory)
        this._watcher.on('change', (eventType, filename) => {
            this._queue.push({ eventType, filename })
        })
        this._watcher.on('error', this.emit.bind(this, 'error'))
        destructible.durable('change', this._change(this._queue.shifter()))
        destructible.destruct(() => this._watcher.close())
        destructible.destruct(() => this._queue.push(null))
    }

    async _change (shifter) {
        for await (const { eventType, filename } of shifter.iterator()) {
            const hash = /-[0-9a-f]{1,8}\.json$/.exec(filename)
            if (hash == null) {
                logger.warn('filename', { directory: this._directory, eventType, filename })
                continue
            }
            let buffer = null
            try {
                buffer = await fs.readFile(path.resolve(this._directory, filename))
            } catch (error) {
                const level = coalesce(LEVEL[error.code], 'warn')
                logger[level]('read', {
                    directory: this._directory,
                    eventType, filename, message: error.message, code: error.code
                })
            }
            try {
                buffer = await fs.readFile(path.resolve(this._directory, filename))
                await fs.unlink(path.resolve(this._directory, filename))
            } catch (error) {
                const level = coalesce(LEVEL[error.code], 'warn')
                logger[level]('unlink', {
                    directory: this._directory,
                    eventType, filename, message: error.message, code: error.code
                })
                continue
            }
            const expected = parseInt(hash, 16)
            const actual = this._hash.call(null, buffer)
            if (actual != expected) {
                logger.warn('checksum', {
                    directory: this._directory, eventType, filename, actual, expected
                })
            }
            const json = buffer.toString()
            try {
                this.emit('data', JSON.parse(json))
            } catch (error) {
                assert(error instanceof SyntaxError, 'syntax error expected')
                logger.warn('json', {
                    directory: this._directory,
                    eventType, filename, json
                })
            }
        }
    }
}

module.exports = Watcher<commit_msg>Isolate JSON parsing into a `try`/`catch` block.<commit_after>const filesystem = require('fs')
const fs = require('fs').promises
const events = require('events')
const Queue = require('avenue')
const path = require('path')
const assert = require('assert')
const logger = require('prolific.logger').createLogger('prolific')
const coalesce = require('extant')

const LEVEL = { ENOENT: 'debug' }

class Watcher extends events.EventEmitter {
    constructor (destructible, hash, directory) {
        super()
        this._queue = new Queue
        this._hash = hash
        this._directory = directory
        this._watcher = filesystem.watch(directory)
        this._watcher.on('change', (eventType, filename) => {
            this._queue.push({ eventType, filename })
        })
        this._watcher.on('error', this.emit.bind(this, 'error'))
        destructible.durable('change', this._change(this._queue.shifter()))
        destructible.destruct(() => this._watcher.close())
        destructible.destruct(() => this._queue.push(null))
    }

    _safeParse (json) {
        try {
            return JSON.parse(json)
        } catch (error) {
            assert(error instanceof SyntaxError, 'syntax error expected')
            logger.warn('json', {
                directory: this._directory,
                eventType, filename, json
            })
        }
    }

    async _change (shifter) {
        for await (const { eventType, filename } of shifter.iterator()) {
            const hash = /-[0-9a-f]{1,8}\.json$/.exec(filename)
            if (hash == null) {
                logger.warn('filename', { directory: this._directory, eventType, filename })
                continue
            }
            let buffer = null
            try {
                buffer = await fs.readFile(path.resolve(this._directory, filename))
            } catch (error) {
                const level = coalesce(LEVEL[error.code], 'warn')
                logger[level]('read', {
                    directory: this._directory,
                    eventType, filename, message: error.message, code: error.code
                })
            }
            try {
                buffer = await fs.readFile(path.resolve(this._directory, filename))
                await fs.unlink(path.resolve(this._directory, filename))
            } catch (error) {
                const level = coalesce(LEVEL[error.code], 'warn')
                logger[level]('unlink', {
                    directory: this._directory,
                    eventType, filename, message: error.message, code: error.code
                })
                continue
            }
            const expected = parseInt(hash, 16)
            const actual = this._hash.call(null, buffer)
            if (actual != expected) {
                logger.warn('checksum', {
                    directory: this._directory, eventType, filename, actual, expected
                })
            }
            const json = this._safeParse(buffer.toString())
            if (json != null) {
                this.emit('data', json)
            }
        }
    }
}

module.exports = Watcher
{'fraction_non_alphanumeric': 0.12082178379259742, 'fraction_numerical': 0.0019566280776129137, 'mean_word_length': 1.7869150386188097, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '496203', 'n_tokens_mistral': 1615, 'n_tokens_neox': 1539, 'n_words': 352}