<commit_before>'use strict'

const path = require('path')
const EventEmitter = require('events').EventEmitter
const mapSeries = require('p-each-series')
const { default: PQueue } = require('p-queue')
const Log = require('ipfs-log')
const Entry = Log.Entry
const Index = require('./Index')
const Replicator = require('./Replicator')
const ReplicationInfo = require('./replication-info')

const Logger = require('logplease')
const logger = Logger.create('orbit-db.store', { color: Logger.Colors.Blue })
Logger.setLogLevel('ERROR')
const io = require('orbit-db-io')

const DefaultOptions = {
  Index: Index,
  maxHistory: -1,
  fetchEntryTimeout: null,
  referenceCount: 32,
  replicationConcurrency: 128,
  syncLocal: false,
  sortFn: undefined
}

class Store {
  constructor (ipfs, identity, address, options) {
    if (!identity) {
      throw new Error('Identity required')
    }

    // Set the options
    const opts = Object.assign({}, DefaultOptions)
    Object.assign(opts, options)
    this.options = opts

    // Default type
    this._type = 'store'

    // Create IDs, names and paths
    this.id = address.toString()
    this.identity = identity
    this.address = address
    this.dbname = address.path || ''
    this.events = new EventEmitter()

    this.remoteHeadsPath = path.join(this.id, '_remoteHeads')
    this.localHeadsPath = path.join(this.id, '_localHeads')
    this.snapshotPath = path.join(this.id, 'snapshot')
    this.queuePath = path.join(this.id, 'queue')
    this.manifestPath = path.join(this.id, '_manifest')

    // External dependencies
    this._ipfs = ipfs
    this._cache = options.cache

    // Access mapping
    const defaultAccess = {
      canAppend: (entry) => (entry.identity.publicKey === identity.publicKey)
    }
    this.access = options.accessController || defaultAccess

    // Create the operations log
    this._oplog = new Log(this._ipfs, this.identity, { logId: this.id, access: this.access, sortFn: this.options.sortFn })

    // _addOperation queue
    this._opqueue = new PQueue({ concurrency: 1 })

    // Create the index
    this._index = new this.options.Index(this.address.root)

    // Replication progress info
    this._replicationStatus = new ReplicationInfo()

    // Statistics
    this._stats = {
      snapshot: {
        bytesLoaded: -1
      },
      syncRequestsReceieved: 0
    }

    try {
      this._replicator = new Replicator(this, this.options.replicationConcurrency)
      // For internal backwards compatibility,
      // to be removed in future releases
      this._loader = this._replicator
      this._replicator.on('load.added', (entry) => {
        // Update the latest entry state (latest is the entry with largest clock time)
        this._replicationStatus.queued++
        this._recalculateReplicationMax(entry.clock ? entry.clock.time : 0)
        // logger.debug(`<replicate>`)
        this.events.emit('replicate', this.address.toString(), entry)
      })
      this._replicator.on('load.progress', (id, hash, entry, have, bufferedLength) => {
        if (this._replicationStatus.buffered > bufferedLength) {
          this._recalculateReplicationProgress(this.replicationStatus.progress + bufferedLength)
        } else {
          this._recalculateReplicationProgress(this._oplog.length + bufferedLength)
        }
        this._replicationStatus.buffered = bufferedLength
        this._recalculateReplicationMax(this.replicationStatus.progress)
        // logger.debug(`<replicate.progress>`)
        this.events.emit('replicate.progress', this.address.toString(), hash, entry, this.replicationStatus.progress, this.replicationStatus.max)
      })

      const onLoadCompleted = async (logs, have) => {
        try {
          for (const log of logs) {
            await this._oplog.join(log)
          }
          this._replicationStatus.queued -= logs.length
          this._replicationStatus.buffered = this._replicator._buffer.length
          await this._updateIndex()

          // only store heads that has been verified and merges
          const heads = this._oplog.heads
          await this._cache.set(this.remoteHeadsPath, heads)
          logger.debug(`Saved heads ${heads.length} [${heads.map(e => e.hash).join(', ')}]`)

          // logger.debug(`<replicated>`)
          this.events.emit('replicated', this.address.toString(), logs.length)
        } catch (e) {
          console.error(e)
        }
      }
      this._replicator.on('load.end', onLoadCompleted)
    } catch (e) {
      console.error('Store Error:', e)
    }
    this.events.on('replicated.progress', (address, hash, entry, progress, have) => {
      this._procEntry(entry)
    })
    this.events.on('write', (address, entry, heads) => {
      this._procEntry(entry)
    })
  }

  get all () {
    return Array.isArray(this._index._index)
      ? this._index._index
      : Object.keys(this._index._index).map(e => this._index._index[e])
  }

  get index () {
    return this._index._index
  }

  get type () {
    return this._type
  }

  get key () {
    return this._key
  }

  /**
   * Returns the database's current replication status information
   * @return {[Object]} [description]
   */
  get replicationStatus () {
    return this._replicationStatus
  }

  setIdentity (identity) {
    this.identity = identity
    this._oplog.setIdentity(identity)
  }

  async close () {
    if (this.options.onClose) {
      await this.options.onClose(this)
    }

    await this._opqueue.onIdle()

    // Replicator teardown logic
    this._replicator.stop()

    // Reset replication statistics
    this._replicationStatus.reset()

    // Reset database statistics
    this._stats = {
      snapshot: {
        bytesLoaded: -1
      },
      syncRequestsReceieved: 0
    }

    // Remove all event listeners<commit_msg>Close access controller on Store.close()<commit_after>'use strict'

const path = require('path')
const EventEmitter = require('events').EventEmitter
const mapSeries = require('p-each-series')
const { default: PQueue } = require('p-queue')
const Log = require('ipfs-log')
const Entry = Log.Entry
const Index = require('./Index')
const Replicator = require('./Replicator')
const ReplicationInfo = require('./replication-info')

const Logger = require('logplease')
const logger = Logger.create('orbit-db.store', { color: Logger.Colors.Blue })
Logger.setLogLevel('ERROR')
const io = require('orbit-db-io')

const DefaultOptions = {
  Index: Index,
  maxHistory: -1,
  fetchEntryTimeout: null,
  referenceCount: 32,
  replicationConcurrency: 128,
  syncLocal: false,
  sortFn: undefined
}

class Store {
  constructor (ipfs, identity, address, options) {
    if (!identity) {
      throw new Error('Identity required')
    }

    // Set the options
    const opts = Object.assign({}, DefaultOptions)
    Object.assign(opts, options)
    this.options = opts

    // Default type
    this._type = 'store'

    // Create IDs, names and paths
    this.id = address.toString()
    this.identity = identity
    this.address = address
    this.dbname = address.path || ''
    this.events = new EventEmitter()

    this.remoteHeadsPath = path.join(this.id, '_remoteHeads')
    this.localHeadsPath = path.join(this.id, '_localHeads')
    this.snapshotPath = path.join(this.id, 'snapshot')
    this.queuePath = path.join(this.id, 'queue')
    this.manifestPath = path.join(this.id, '_manifest')

    // External dependencies
    this._ipfs = ipfs
    this._cache = options.cache

    // Access mapping
    const defaultAccess = {
      canAppend: (entry) => (entry.identity.publicKey === identity.publicKey)
    }
    this.access = options.accessController || defaultAccess

    // Create the operations log
    this._oplog = new Log(this._ipfs, this.identity, { logId: this.id, access: this.access, sortFn: this.options.sortFn })

    // _addOperation queue
    this._opqueue = new PQueue({ concurrency: 1 })

    // Create the index
    this._index = new this.options.Index(this.address.root)

    // Replication progress info
    this._replicationStatus = new ReplicationInfo()

    // Statistics
    this._stats = {
      snapshot: {
        bytesLoaded: -1
      },
      syncRequestsReceieved: 0
    }

    try {
      this._replicator = new Replicator(this, this.options.replicationConcurrency)
      // For internal backwards compatibility,
      // to be removed in future releases
      this._loader = this._replicator
      this._replicator.on('load.added', (entry) => {
        // Update the latest entry state (latest is the entry with largest clock time)
        this._replicationStatus.queued++
        this._recalculateReplicationMax(entry.clock ? entry.clock.time : 0)
        // logger.debug(`<replicate>`)
        this.events.emit('replicate', this.address.toString(), entry)
      })
      this._replicator.on('load.progress', (id, hash, entry, have, bufferedLength) => {
        if (this._replicationStatus.buffered > bufferedLength) {
          this._recalculateReplicationProgress(this.replicationStatus.progress + bufferedLength)
        } else {
          this._recalculateReplicationProgress(this._oplog.length + bufferedLength)
        }
        this._replicationStatus.buffered = bufferedLength
        this._recalculateReplicationMax(this.replicationStatus.progress)
        // logger.debug(`<replicate.progress>`)
        this.events.emit('replicate.progress', this.address.toString(), hash, entry, this.replicationStatus.progress, this.replicationStatus.max)
      })

      const onLoadCompleted = async (logs, have) => {
        try {
          for (const log of logs) {
            await this._oplog.join(log)
          }
          this._replicationStatus.queued -= logs.length
          this._replicationStatus.buffered = this._replicator._buffer.length
          await this._updateIndex()

          // only store heads that has been verified and merges
          const heads = this._oplog.heads
          await this._cache.set(this.remoteHeadsPath, heads)
          logger.debug(`Saved heads ${heads.length} [${heads.map(e => e.hash).join(', ')}]`)

          // logger.debug(`<replicated>`)
          this.events.emit('replicated', this.address.toString(), logs.length)
        } catch (e) {
          console.error(e)
        }
      }
      this._replicator.on('load.end', onLoadCompleted)
    } catch (e) {
      console.error('Store Error:', e)
    }
    this.events.on('replicated.progress', (address, hash, entry, progress, have) => {
      this._procEntry(entry)
    })
    this.events.on('write', (address, entry, heads) => {
      this._procEntry(entry)
    })
  }

  get all () {
    return Array.isArray(this._index._index)
      ? this._index._index
      : Object.keys(this._index._index).map(e => this._index._index[e])
  }

  get index () {
    return this._index._index
  }

  get type () {
    return this._type
  }

  get key () {
    return this._key
  }

  /**
   * Returns the database's current replication status information
   * @return {[Object]} [description]
   */
  get replicationStatus () {
    return this._replicationStatus
  }

  setIdentity (identity) {
    this.identity = identity
    this._oplog.setIdentity(identity)
  }

  async close () {
    if (this.options.onClose) {
      await this.options.onClose(this)
    }

    await this._opqueue.onIdle()

    // Replicator teardown logic
    this._replicator.stop()

    // Reset replication statistics
    this._replicationStatus.reset()

    // Reset database statistics
    this._stats = {
      snapshot: {
        bytesLoaded: -1
      },
      syncRequestsReceieved: 0
    }

    // Close store access controller
    await this.access.close()

    // Remove all event listeners
{'fraction_non_alphanumeric': 0.12447600307981864, 'fraction_numerical': 0.0020532124219351526, 'mean_word_length': 2.7831715210355985, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 27, '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': '563206', 'n_tokens_mistral': 3483, 'n_tokens_neox': 3404, 'n_words': 749}