index.js

const fs = require('fs')
const EventEmitter = require('events')
const getDirName = require('path').dirname
const path = require('path')
const Database = require('./lib/database')
const Hyperbee = require('hyperbee')
const Hypercore = require('./lib/core')
const pump = require('pump')
const Crypto = require('./lib/crypto')
const Swarm = require('./lib/swarm')
const stream = require('stream')
const blake = require('blakejs')
const Hyperswarm = require('hyperswarm')
const DHT = require('@hyperswarm/dht')
const HyperbeeMessages = require('hyperbee/lib/messages.js')
const MemoryStream = require('memorystream')
const { v4: uuidv4 } = require('uuid')
const FixedChunker = require('./util/fixedChunker.js')
const RequestChunker = require('./util/requestChunker.js')
const WorkerKeyPairs = require('./util/workerKeyPairs.js')
const isOnline = require('is-online')
const BSON = require('bson')
const FileDB = require('./util/filedb.util')

const HASH_OUTPUT_LENGTH = 32 // bytes
const MAX_PLAINTEXT_BLOCK_SIZE = 65536
const MAX_ENCRYPTED_BLOCK_SIZE = 65553
const FILE_TIMEOUT = 10000 // How long to wait for the on data event when downloading a file from a remote drive.
const FILE_RETRY_ATTEMPTS = 2 // Fail to fetch file after 3 attempts
const FILE_BATCH_SIZE = 10 // How many parallel requests are made in each file request batch


class Drive extends EventEmitter {
  constructor(
    drivePath, 
    peerPubKey, // Key used to clone and seed drive. Should only be shared with trusted sources
    { 
      storage, 
      keyPair, // ed25519 keypair to listen on
      writable, 
      swarmOpts, 
      encryptionKey, 
      fileTimeout, 
      fileRetryAttempts, 
      checkNetworkStatus, 
      joinSwarm,
      fullTextSearch, // Initialize a corestore to support full text search indexes.
      blind, // Set to true if blind mirroring another drive (you don't have the encryption key)
      storageMaxBytes // Max size this drive will store in bytes before turning off replication/file syncing
    }
  ) {
    super()

    this.storage = storage
    this.encryptionKey = encryptionKey
    this.database = null;
    this.db = null;
    this.drivePath = drivePath
    this.swarmOpts = swarmOpts
    this.publicKey = null
    this.peerPubKey = peerPubKey 
    this.keyPair = keyPair ? keyPair : DHT.keyPair()
    this.writable = writable
    this.fullTextSearch = fullTextSearch 
    this.fileTimeout = fileTimeout || FILE_TIMEOUT
    this.fileRetryAttempts = fileRetryAttempts-1 || FILE_RETRY_ATTEMPTS-1
    this.requestQueue = new RequestChunker(null, FILE_BATCH_SIZE)
    this.checkNetworkStatus = checkNetworkStatus
    this.joinSwarm = typeof joinSwarm === 'boolean' ? joinSwarm : true
    this.network = {
      internet: false,
      drive: false
    }
    this.blind = blind ? blind : false
    this.storageMaxBytes = storageMaxBytes || Infinity

    // When using custom storage, transform drive path into beginning of the storage namespace
    this.storageName = drivePath.slice(drivePath.lastIndexOf('/') + 1, drivePath.length)
  
    this._localCore = null
    this._localHB = null // Optional datastore for storing encrypted data locally. This will not sync with peers or be replicated.
    this._swarm = null
    this._workerKeyPairs = new WorkerKeyPairs(FILE_BATCH_SIZE)
    this._collections = {}
    this._filesDir = path.join(drivePath, `./Files`)
    this._localDB = null // Local Key value datastore only
    this._lastSeq = null
    this._checkInternetInt = null
    this._checkInternetInProgress = false
    this._fileStatPath = this.drivePath + '/Files/file_stat.txt'
    this._stat = {
      file_bytes: 0,
      core_bytes: 0,
      total_bytes: 0
    }

    if (!fs.existsSync(drivePath)) {
      fs.mkdirSync(drivePath)
    }

    if (!fs.existsSync(this._filesDir)) {
      fs.mkdirSync(this._filesDir)
    }

    this.requestQueue.on('process-queue', async files => {
      this.requestQueue.reset()

      await this.fetchFileBatch(files, (stream, file) => {
        return new Promise((resolve, reject) => {
          fs.mkdirSync(getDirName(this._filesDir + file.path), { recursive: true })

          const writeStream = fs.createWriteStream(this._filesDir + file.path)

          pump(stream, writeStream, (err) => {
            if (err) reject(err)

            setTimeout(() => {
              this.emit('file-sync', file)
            })

            resolve()
          })
        })
      })
    })

    process.on('uncaughtException', err => {
      // gracefully catch uncaught exceptions
    })

    // Periodically check this drive's connection to the internet.
    // When the internet is down, emit a network status updated event.
    if(this.checkNetworkStatus) {
      this._checkInternetInt = setInterval(async () => {
        if(!this._checkInternetInProgress) {
          this._checkInternetInProgress = true
          await this._checkInternet();
          this._checkInternetInProgress = false
        }
      }, 1500)
    }
  }

  async ready() {
    await this._bootstrap()

    const stat = await this._localDB.get('stat')

    if(!stat) {
      await this.database._updateStatBytes(0)
      await this._localDB.put('stat', { ...this._stat })
    } else {
      this._stat = stat
    }

    this.publicKey = this.database.localMetaCore.key.toString('hex')

    if (this.peerPubKey) {
      this.discoveryKey = createTopicHash(this.peerPubKey).toString('hex')
    } else {
      this.discoveryKey = createTopicHash(this.publicKey).toString('hex')
    }

    // Data here can only be read by peer drives
    // that are sharing the same drive secret
    this._collections.files = await this.database.collection('file')

    // This drastically speeds up queries and is necessary for sorting by fields
    await this._collections.files.createIndex(['path'])

    if (this.keyPair && this.joinSwarm) {
      await this.connect()
    }

    const stream = this.metadb.createReadStream({ live: true })
    
    stream.on('data', async data => {
      if(data.value.toString().indexOf('hyperbee') === -1) {
        const op = HyperbeeMessages.Node.decode(data.value)
        const node = {
          key: op.key.toString('utf8'),
          value: JSON.parse(op.value.toString('utf8')),
          seq: data.seq
        }
 
        if (node.key !== '__peers') {
          await this._update(node)
        }
      }
    })

    if(!this.blind) {
      const lastCollectionSeq = await this._localDB.get('collection-lastSeq')

      const cStream = this.database.autobee.createReadStream({ live: true }) // Collections stream

      cStream.on('data', async data => {
        if(data.value.toString().indexOf('hyperbee') === -1) {

          if(lastCollectionSeq && lastCollectionSeq.seq && data.seq < lastCollectionSeq.seq) return
          
          await this._localDB.put('collection-lastSeq', { seq: data.seq })
          const op = HyperbeeMessages.Node.decode(data.value)
          const key = op.key.toString('utf8')
          let collection
          let value

          if(key.split('-').length === 1) return

          collection = key.split('-')[0]

          try {
            value = BSON.deserialize(op.value)
          } catch(err) {
            // womp womp
          }

          if(value && value._id) {
            const node = {
              collection,
              _id: value._id.toString('hex'),
              author: value.author
            }
            this.emit('collection-update', node)
          }
        }
      })
    } else {
      // If this drive can't decipher the data inside the remote hypercore's then just listen for when those cores are updated.
      this.database.on('collection-update', () => {
        this.emit('collection-update')
      })
    }

    this.opened = true
  }

  // Connect to the Hyperswarm network
  async connect() {
    if (this._swarm) {
      await this._swarm.close()
    }

    this._swarm = new Swarm({
      keyPair: this.keyPair,
      blind: this.blind,
      workerKeyPairs: this._workerKeyPairs.keyPairs,
      topic: this.discoveryKey,
      publicKey: this.peerPubKey || this.publicKey,
      isServer: this.swarmOpts.server,
      isClient: this.swarmOpts.client,
      acl: this.swarmOpts.acl
    })

    if(this.checkNetworkStatus) {
      this._swarm.on('disconnected', () => {
        if(this.network.drive) {
          this.network.drive = false
          this.emit('network-updated', { drive: this.network.drive })
        }
      })

      this._swarm.on('connected', () => {
        if(!this.network.drive) {
          this.network.drive = true
          this.emit('network-updated', { drive: this.network.drive })
        }
      })
    }

    if(!this.blind) {
      this._swarm.on('message', (peerPubKey, data) => {
        this.emit('message', peerPubKey, data)
      })
    }
    
    this._swarm.on('file-requested', socket => {
      socket.once('data', async data => {
        const fileHash = data.toString('utf-8')
        const file = await this.metadb.get(fileHash)
      
        if (!file || file.value.deleted) {
          let err = new Error()
          err.message = 'Requested file was not found on drive'
          socket.destroy(err)
        } else {
          const readStream = fs.createReadStream(path.join(this.drivePath, `./Files${file.value.path}`))
          pump(readStream, socket, (err) => {
            // handle done
          })
        }
      })
      
      socket.on('error', (err) => {
        // handle errors
      })
    })

    this._swarm.ready()
  }

  async addPeer(peerKey) {
    const remotePeers = await this._localDB.get('remotePeers')

    const peers = [...remotePeers.value, peerKey]

    await this._localDB.put('remotePeers', { ...peers })

    await this.database.addInput(peerKey)
  }

  // Remove Peer
  async removePeer(peerKey) {
    let remotePeers = await this._localDB.get('remotePeers')

    const peers = remotePeers.filter(peer => peer !== peerKey)

    await this._localDB.put('remotePeers', { ...peers })

    await this.database.removeInput(peerKey)
  }

  async writeFile(path, readStream, opts = {}) {
    let filePath = path
    let dest
    const uuid = uuidv4()

    if (filePath[0] === '/') {
      filePath = filePath.slice(1, filePath.length)
    }

    if (opts.encrypted) {
      dest = `${this._filesDir}/${uuid}`
    } else {
      fs.mkdirSync(getDirName(this._filesDir + path), { recursive: true })
      dest = this._filesDir + path
    }

    return new Promise(async (resolve, reject) => {
      const pathSeg = filePath.split('/')
      let fullFile = pathSeg[pathSeg.length - 1]
      let fileName
      let fileExt

      if (fullFile.indexOf('.') > -1) {
        fileName = fullFile.split('.')[0]
        fileExt = fullFile.split('.')[1]
      }

      const writeStream = fs.createWriteStream(dest)

      if (opts.encrypted && !opts.skipEncryption) {
        const fixedChunker = new FixedChunker(readStream, MAX_PLAINTEXT_BLOCK_SIZE)
        const { key, header, file } = await Crypto.encryptStream(fixedChunker, writeStream)

        await this.database._updateStatBytes(file.size)

        await this.metadb.put(file.hash, {
          uuid,
          size: file.size,
          hash: file.hash,
          path: `/${uuid}`,
          peer_key: this.keyPair.publicKey.toString('hex'),
          discovery_key: this.discoveryKey
        })

        const fileMeta = {
          uuid,
          name: fileName,
          size: file.size,
          mimetype: fileExt,
          encrypted: true,
          key: key.toString('hex'),
          header: header.toString('hex'),
          hash: file.hash,
          path: filePath,
          peer_key: this.keyPair.publicKey.toString('hex'),
          discovery_key: this.discoveryKey
        }

        await this._collections.files.insert({ ...fileMeta, updatedAt: new Date().toISOString() })

        this.emit('file-add', fileMeta)

        resolve({
          key: key.toString('hex'),
          header: header.toString('hex'),
          ...fileMeta
        })
      } else {
        let bytes = 0
        const hash = blake.blake2bInit(HASH_OUTPUT_LENGTH, null)
        const calcHash = new stream.Transform({
          transform
        })

        function transform(chunk, encoding, callback) {
          bytes += chunk.byteLength

          blake.blake2bUpdate(hash, chunk)
          callback(null, chunk)
        }

        pump(readStream, calcHash, writeStream, async () => {
          setTimeout(async () => {
            const _hash = Buffer.from(blake.blake2bFinal(hash)).toString('hex')

            if (bytes > 0) {
              await this.database._updateStatBytes(bytes)

              await this.metadb.put(_hash, {
                uuid,
                size: bytes,
                hash: _hash,
                path,
                peer_key: this.keyPair.publicKey.toString('hex'),
                discovery_key: this.discoveryKey
              })

              const fileMeta = {
                uuid,
                name: fileName,
                size: bytes,
                mimetype: fileExt,
                hash: _hash,
                path: filePath,
                peer_key: this.keyPair.publicKey.toString('hex'),
                discovery_key: this.discoveryKey
              }

              await this._collections.files.insert({ ...fileMeta, updatedAt: new Date().toISOString() })

              this.emit('file-add', fileMeta)
              resolve(fileMeta)
            } else {
              reject('No bytes were written.')
            }
          })
        })
      }
    })
  }

  async readFile(path) {
    let file
    let filePath = path

    if (filePath[0] === '/') {
      filePath = filePath.slice(1, filePath.length)
    }

    try {
      file = await this._collections.files.findOne({ path: filePath })

      const stream = fs.createReadStream(`${this._filesDir}/${file.uuid}`)

      // If key then decipher file
      if (file.encrypted && file.key && file.header) {
        const fixedChunker = new FixedChunker(stream, MAX_ENCRYPTED_BLOCK_SIZE)
        return Crypto.decryptStream(fixedChunker, file.key, file.header)
      } else {
        return stream
      }
    } catch (err) {
      throw err
    }
  }

  decryptFileStream(stream, key, header) {
    const fixedChunker = new FixedChunker(stream, MAX_ENCRYPTED_BLOCK_SIZE)
    return Crypto.decryptStream(fixedChunker, key, header)
  }

  // TODO: Implement this
  fetchFileByHash(fileHash) {
  }

  async fetchFileByDriveHash(discoveryKey, fileHash, opts = {}) {
    const keyPair = opts.keyPair || this.keyPair
    const memStream = new MemoryStream()
    const topic = blake.blake2bHex(discoveryKey, null, HASH_OUTPUT_LENGTH)


    if (!fileHash || typeof fileHash !== 'string') {
      return reject('File hash is required before making a request.')
    }

    if (!discoveryKey || typeof discoveryKey !== 'string') {
      return reject('Discovery key cannot be null and must be a string.')
    }

    try {
      await this._initFileSwarm(memStream, topic, fileHash, 0, { keyPair })
    } catch(e) {      
      setTimeout(() => {
        memStream.destroy(e)
      })
      return memStream
    }

    if (opts.key && opts.header) {
      return this.decryptFileStream(memStream, opts.key, opts.header)
    }

    return memStream
  }

  async fetchFileBatch(files, cb) {
    const batches = new RequestChunker(files, FILE_BATCH_SIZE)

    for (let batch of batches) {
      const requests = []

      for (let file of batch) {
        
        if(typeof file.size === 'number') await this.database._updateStatBytes(file.size)
        const stat = await this._localDB.get('stat')
        if(stat.total_bytes <= this.storageMaxBytes) {
          requests.push(new Promise(async (resolve, reject) => {
            if (file.discovery_key) {
              const keyPair = this._workerKeyPairs.getKeyPair()
              const stream = await this.fetchFileByDriveHash(file.discovery_key, file.hash, { key: file.key, header: file.header, keyPair })

              await cb(stream, file)

              resolve()
            } else {
              // TODO: Fetch files by hash
            }
          }))
        }
      }

      await Promise.all(requests)
      this.requestQueue.queue = []
    }
  }

  async _initFileSwarm(stream, topic, fileHash, attempts, { keyPair }) {
    return new Promise((resolve, reject) => {
      if (attempts > this.fileRetryAttempts) {
        const err = new Error('Unable to make a connection or receive data within the allotted time.')
        err.fileHash = fileHash
        this._workerKeyPairs.release(keyPair.publicKey.toString('hex'))
        stream.destroy(err)
        return reject(err)
      }

      const swarm = new Hyperswarm({ keyPair })

      let connected = false
      let receivedData = false
      let streamError = false

      swarm.join(Buffer.from(topic, 'hex'), { server: false, client: true })

      swarm.on('connection', async (socket, info) => {
        receivedData = false

        if (!connected) {
          connected = true

          // Tell the host drive which file we want
          socket.write(fileHash)

          socket.on('data', (data) => {
            resolve()
            stream.write(data)
            receivedData = true
          })

          socket.once('end', () => {
            if (receivedData) {
              this._workerKeyPairs.release(keyPair.publicKey.toString('hex'))
              stream.end()
              swarm.destroy()
            }
          })

          socket.once('error', (err) => {
            stream.destroy(err)
            streamError = true
            reject(err)
          })
        }
      })

      setTimeout(async () => {
        if (!connected || streamError || !receivedData) {
          attempts += 1
          await swarm.leave(topic)
          await swarm.destroy()

          try {
            await this._initFileSwarm(stream, topic, fileHash, attempts, { keyPair })
            resolve()
          } catch(e) {
            reject(e)
          }
        }
      }, this.fileTimeout)
    })
  }

  async _checkInternet() {
    return new Promise((resolve, reject) => {
      isOnline().then((isOnline) => {
        if(!isOnline && this.network.internet) {
          this.network.internet = false
          this.emit('network-updated', { internet: this.network.internet })
        }

        if(isOnline && !this.network.internet) {
          this.network.internet = true
          this.emit('network-updated', { internet: this.network.internet })
        }

        resolve()
      })
    })
  }

  async unlink(filePath) {
    let fp = filePath

    if (fp[0] === '/') {
      fp = filePath.slice(1, fp.length)
    }

    try {
      const _file = await this._collections.files.findOne({ path: fp })

      if (!_file) return

      const file = await this.metadb.get(_file.hash)

      if(!file) return

      fs.unlinkSync(path.join(this._filesDir, file.value.path))

      await this._collections.files.update({ _id: _file._id} , { uuid: file.value.uuid, deleted: true, updatedAt: new Date().toISOString() })

      await this.database._updateStatBytes(-Math.abs(file.value.size))

      await this.metadb.put(file.value.hash, {
        uuid: file.value.uuid,
        discovery_key: file.value.discovery_key,
        size: file.value.size,
        deleted: true
      })

      this.emit('file-unlink', file.value)
    } catch (err) {
      throw err
    }
  }

  async destroyHyperfile(path) {
    const filePath = await this.bee.get(path)
    const file = await this.bee.get(filePath.value.hash)
    await this._clearStorage(file.value)
  }

  async _bootstrap() {
    if(!this.blind) {
      this._localCore = new Hypercore(path.join(this.drivePath, `./LocalCore`), { storageNamespace: `${this.storageName}:local-core` })
      await this._localCore.ready()
      this._localHB = new Hyperbee(this._localCore, { keyEncoding: 'utf8', valueEncoding: 'json' })
    }
    
    this._localDB = new FileDB(`${this.drivePath}/LocalDS`)

    this.database = new Database(this.storage || this.drivePath, {
      localDB: this._localDB,
      keyPair: this.keyPair,
      storageName: this.storageName,
      encryptionKey: this.encryptionKey,
      peerPubKey: this.peerPubKey,
      acl: this.swarmOpts && this.swarmOpts.acl ? this.swarmOpts.acl : null,
      joinSwarm: this.joinSwarm,
      fts: this.fullTextSearch,
      blind: this.blind,
      stat: this._stat,
      storageMaxBytes: this.storageMaxBytes,
      fileStatPath: this._fileStatPath
    })

    this.database.on('disconnected', () => {
      if(this.network.drive) {
        this.network.drive = false
        this.emit('network-updated', { drive: this.network.drive })
      }
    })

    this.database.on('connected', () => {
      if(!this.network.drive) {
        this.network.drive = true
        this.emit('network-updated', { drive: this.network.drive })
      }
    })

    if(this.checkNetworkStatus) {
      this.database.on('disconnected', () => {
        if(this.network.drive) {
          this.network.drive = false
          this.emit('network-updated', { drive: this.network.drive })
        }
      })
    }

    await this.database.ready()

    this.db = this.database
    this.metadb = this.database.metadb
  }

  async _update(data) {
    let lastMetaSeq

    lastMetaSeq = await this._localDB.get('meta-lastSeq')

    if (!lastMetaSeq || !lastMetaSeq.seq) lastMetaSeq = { seq: null }
    
    this.emit('sync')

    if (
      !data.value.deleted &&
      data.value.peer_key !== this.keyPair.publicKey.toString('hex') &&
      lastMetaSeq.seq !== data.seq
    ) {
      if (data.value.hash) {
        try {
          await this._localDB.put('meta-lastSeq', { seq: data.seq })
          const stat = await this._localDB.get('stat')
          
          if(stat.total_bytes <= this.storageMaxBytes) {  
            this.requestQueue.addFile(data.value)
          }
        } catch (err) {
          throw err
        }
      }
    }

    if (
      data.value.deleted &&
      data.value.peer_key !== this.keyPair.publicKey.toString('hex')
    ) {
      try {
        const filePath = path.join(this._filesDir, `/${data.value.uuid}`)
        if (fs.existsSync(filePath)) {
          fs.unlinkSync(filePath)

          await this.database._updateStatBytes(-Math.abs(data.value.size))

          setTimeout(() => {
            this.emit('file-unlink', data.value)
          })
        }
      } catch (err) {
        throw err
      }
    }
  }

  // Deprecated
  info() {
    const bytes = getTotalSize(this.drivePath)
    return {
      size: bytes
    }
  }

  async stat() {
    return this._localDB.get('stat')
  }

  /**
   * Close drive and disconnect from all Hyperswarm topics
   */
  async close() {
    if(this.joinSwarm) {
      await this._swarm.close()
    }

    if(this._localCore) {
      await this._localCore.close()
    }

    await this.database.close()

    clearInterval(this._checkInternetInt)

    this.network = {
      internet: false,
      drive: false
    }

    this.emit('network-updated', this.network)

    if(this.checkNetworkStatus) {
      clearInterval(this._checkInternetInt)

      this.network = {
        internet: false,
        drive: false
      }

      this.emit('network-updated', this.network)
    }

    this.openend = false
  }
}

function createTopicHash(topic) {
  const crypto = require('crypto')

  return crypto.createHash('sha256')
    .update(topic)
    .digest()
}

async function auditFile(stream, remoteHash) {
  return new Promise((resolve, reject) => {
    let hash = blake.blake2bInit(HASH_OUTPUT_LENGTH, null)

    stream.on('error', err => reject(err))
    stream.on('data', chunk => {
      blake.blake2bUpdate(hash, chunk)
    })
    stream.on('end', () => {
      const localHash = Buffer.from(blake.blake2bFinal(hash)).toString('hex')

      if (localHash === remoteHash)
        return resolve()

      reject('Hashes do not match')
    })
  })
}


const getAllFiles = function (dirPath, arrayOfFiles) {
  files = fs.readdirSync(dirPath)

  arrayOfFiles = arrayOfFiles || []

  files.forEach(function (file) {
    if (fs.statSync(dirPath + "/" + file).isDirectory()) {
      arrayOfFiles = getAllFiles(dirPath + "/" + file, arrayOfFiles)
    } else {
      arrayOfFiles.push(path.join(dirPath, file))
    }
  })

  return arrayOfFiles
}

const getTotalSize = function (directoryPath) {
  const arrayOfFiles = getAllFiles(directoryPath)

  let totalSize = 0

  arrayOfFiles.forEach(function (filePath) {
    totalSize += fs.statSync(filePath).size
  })

  return totalSize
}

module.exports = Drive