<commit_before>'use strict'

var util = require('util')
var Readable = require('stream').Readable
var crypto = require('crypto')
var co = require('co')
// var Promise = require('bluebird')

var noErr = function (err) { err }

var _downloadFilename = function (ctx, filename, options) {
  Readable.call(this)
  this.options = Object.assign({
    revision: -1,
    seekStart: null,
    seekEnd: null
  }, options)
  this.conf = ctx.conf
  this.r = ctx.r
  this.filename = filename
  this.hash = crypto.createHash('sha256')
  this.num = 0
}

util.inherits(_downloadFilename, Readable)

_downloadFilename.prototype._read = function () {
  co(function *() {
    let chunk
    let startNum = 0
    let startOffset = 0
    let endNum = null
    let endOffset = null
    let verifyHash = (this.options.seekStart == null && this.options.seekEnd == null)

    if (!this.file) {
      let revSteps
      let query = this.r.table(this.conf.fileTable).between(['Complete', this.filename, this.r.minval], ['Complete', this.filename, this.r.maxval], {index: 'file_ix'})
      if (this.options.revision >= 0) {
        revSteps = this.options.revision
        query = query.orderBy({index: this.r.asc('file_ix')})
      } else {
        revSteps = (this.options.revision * -1) - 1
        query = query.orderBy({index: this.r.desc('file_ix')})
      }

      let files = yield query.run()

      if (files.length === 0) { throw new Error('File not found!') }
      if (files.length < (revSteps + 1)) { throw new Error('File revision does not exist!') }

      this.file = files[revSteps]
      this.chunkSize = this.file.chunkSizeBytes
      if (this.options.seekStart != null) {
        startNum = Math.floor(this.options.seekStart / this.chunkSize)
        startOffset = this.options.seekStart % this.chunkSize
      }
      if (this.options.seekEnd != null) {
        endNum = Math.floor((this.options.seekEnd) / this.chunkSize)
        endOffset = (this.options.seekEnd % this.chunkSize) + 1
      }
      this.num = startNum
      if (endNum != null) {
        this.cursor = yield this.r.table(this.conf.chunkTable).between([this.file.id, startNum], [this.file.id, endNum], {index: 'chunk_ix', rightBound: 'closed'}).orderBy({index: 'chunk_ix'}).run({cursor: true})
      } else {
        this.cursor = yield this.r.table(this.conf.chunkTable).between([this.file.id, startNum], [this.file.id, this.r.maxval], {index: 'chunk_ix', rightBound: 'closed'}).orderBy({index: 'chunk_ix'}).run({cursor: true})
      }
    }

    try {
      chunk = yield this.cursor.next()
    } catch (err) {
      if (err.message !== 'No more rows in the cursor.') {
        throw err
      }
    }

    if (chunk) {
      if (verifyHash) { this.hash.update(chunk.data) }
      if (this.num === startNum && this.num === endNum) {
        this.push(chunk.data.slice(startOffset, endOffset))
      } else if (this.num === startNum) {
        this.push(chunk.data.slice(startOffset))
      } else if (this.num === endNum) {
        this.push(chunk.data.slice(0, endOffset))
      } else {
        this.push(chunk.data)
      }<commit_msg>verify chunk numbers<commit_after>'use strict'

var util = require('util')
var Readable = require('stream').Readable
var crypto = require('crypto')
var co = require('co')
// var Promise = require('bluebird')

var noErr = function (err) { err }

var _downloadFilename = function (ctx, filename, options) {
  Readable.call(this)
  this.options = Object.assign({
    revision: -1,
    seekStart: null,
    seekEnd: null
  }, options)
  this.conf = ctx.conf
  this.r = ctx.r
  this.filename = filename
  this.hash = crypto.createHash('sha256')
  this.num = 0
}

util.inherits(_downloadFilename, Readable)

_downloadFilename.prototype._read = function () {
  co(function *() {
    let chunk
    let startNum = 0
    let startOffset = 0
    let endNum = null
    let endOffset = null
    let verifyHash = (this.options.seekStart == null && this.options.seekEnd == null)

    if (!this.file) {
      let revSteps
      let query = this.r.table(this.conf.fileTable).between(['Complete', this.filename, this.r.minval], ['Complete', this.filename, this.r.maxval], {index: 'file_ix'})
      if (this.options.revision >= 0) {
        revSteps = this.options.revision
        query = query.orderBy({index: this.r.asc('file_ix')})
      } else {
        revSteps = (this.options.revision * -1) - 1
        query = query.orderBy({index: this.r.desc('file_ix')})
      }

      let files = yield query.run()

      if (files.length === 0) { throw new Error('File not found!') }
      if (files.length < (revSteps + 1)) { throw new Error('File revision does not exist!') }

      this.file = files[revSteps]
      this.chunkSize = this.file.chunkSizeBytes
      if (this.options.seekStart != null) {
        startNum = Math.floor(this.options.seekStart / this.chunkSize)
        startOffset = this.options.seekStart % this.chunkSize
      }
      if (this.options.seekEnd != null) {
        endNum = Math.floor((this.options.seekEnd) / this.chunkSize)
        endOffset = (this.options.seekEnd % this.chunkSize) + 1
      }
      this.num = startNum
      if (endNum != null) {
        this.cursor = yield this.r.table(this.conf.chunkTable).between([this.file.id, startNum], [this.file.id, endNum], {index: 'chunk_ix', rightBound: 'closed'}).orderBy({index: 'chunk_ix'}).run({cursor: true})
      } else {
        this.cursor = yield this.r.table(this.conf.chunkTable).between([this.file.id, startNum], [this.file.id, this.r.maxval], {index: 'chunk_ix', rightBound: 'closed'}).orderBy({index: 'chunk_ix'}).run({cursor: true})
      }
    }

    try {
      chunk = yield this.cursor.next()
    } catch (err) {
      if (err.message !== 'No more rows in the cursor.') {
        throw err
      }
    }

    if (chunk) {
      if (chunk.num !== this.num) {
        throw new Error('Chunk number mismatch: File is likely corrupted!')
      }
      if (verifyHash) { this.hash.update(chunk.data) }
      if (this.num === startNum && this.num === endNum) {
        this.push(chunk.data.slice(startOffset, endOffset))
      } else if (this.num === startNum) {
        this.push(chunk.data.slice(startOffset))
      } else if (this.num === endNum) {
        this.push(chunk.data.slice(0, endOffset))
      } else {
        this.push(chunk.data)
      }
{'fraction_non_alphanumeric': 0.15468478432611787, 'fraction_numerical': 0.004424079633433402, 'mean_word_length': 2.781362007168459, 'pattern_counts': {'":': 0, '<': 5, '<?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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2814294', 'n_tokens_mistral': 2081, 'n_tokens_neox': 2047, 'n_words': 454}