replicate-peers.js

/*
Copyright 2017 The BioBricks Foundation

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */

var AJV = require('ajv')
var attachmentPath = require('./util/attachment-path')
var concat = require('concat-stream')
var encoding = require('./encoding')
var flushWriteStream = require('flush-write-stream')
var fs = require('fs')
var http = require('http-https')
var latest = require('./latest')
var mkdirp = require('mkdirp')
var parse = require('json-parse-errback')
var path = require('path')
var pump = require('pump')
var readRecord = require('./util/read-record')
var recordDirectoryPath = require('./util/record-directory-path')
var recordPath = require('./util/record-path')
var runParallel = require('run-parallel')
var runSeries = require('run-series')
var signaturePath = require('./util/timestamp-path')
var sodium = require('sodium-prebuilt').api
var split2 = require('split2')
var stringify = require('json-stable-stringify')
var through2 = require('through2')
var timestampPath = require('./util/timestamp-path')
var url = require('url')
var xtend = require('xtend')

var timestampSchema = latest(require('./schemas/timestamp'))

var validatePublication = new AJV({allErrors: true})
  .compile(latest(require('./schemas/publication')))

var validateTimestamp = new AJV({allErrors: true})
  .compile(timestampSchema)

module.exports = function (configuration, log) {
  var directory = configuration.directory
  readPeers(directory, log, function (error, peers) {
    if (error) return logError(error)
    log.info({
      peers: peers.map(function (peer) {
        return url.format(peer.url)
      })
    }, 'read peers')
    runParallel(
      peers.map(function (peer) {
        return function (done) {
          var peerLog = log.child({peer: peer.url.hostname})
          peerLog.info('replicating')
          replicatePeer(configuration, peerLog, peer, done)
        }
      }),
      function (error) {
        if (error) return logError(error)
        if (peers.length === 0) {
          log.info('done')
        } else {
          writePeers(directory, peers, function (error) {
            if (error) return logError(error)
            log.info('done')
          })
        }
      }
    )
  })

  function logError (error) {
    log.error(error)
  }
}

function replicatePeer (configuration, log, peer, done) {
  var request = xtend(peer.url, {
    path: peer.url.path + '/accessions?from=' + (peer.last + 1),
    headers: {accept: 'text/csv'}
  })
  http.request(request)
    .once('error', function (error) {
      done(error)
    })
    .once('response', function (response) {
      var counter = peer.last
      pump(
        response,
        split2(),
        through2.obj(function (line, _, done) {
          counter++
          done(null, {
            accessionNumber: counter,
            digest: line.toString().split(',')[1]
          })
        }),
        flushWriteStream.obj(function (publication, _, done) {
          var recordLog = log.child({digest: publication.digest})
          republish(configuration, recordLog, peer, publication, done)
        }),
        function (error) {
          if (error) {
            done(error)
          } else {
            done()
          }
        }
      )
    })
    .end()
}

function republish (configuration, log, peer, publication, done) {
  var time = new Date().toISOString()
  var files = []
  var haveRecord
  var havePeerTimestamp
  var record
  var peerTimestamp
  runSeries([
    runParallel.bind(null, [
      checkHaveRecord,
      checkHavePeerTimestamp
    ]),
    runParallel.bind(null, [
      getRecord,
      unless(havePeerTimestamp, getPeerTimestamp)
    ]),
    unless(haveRecord, validateRecord),
    unless(havePeerTimestamp, validatePeerTimestamp),
    unless(haveRecord, makeDirectory),
    runParallel.bind(null, [
      unless(haveRecord, saveOwnTimestamp),
      unless(havePeerTimestamp, savePeerTimestamp),
      unless(haveRecord, saveRecord),
      unless(haveRecord, getAndSaveAttachments)
    ]),
    unless(haveRecord, appendToAccessions),
    bumpAccessionNumber
  ], function (error) {
    if (error) {
      log.error(error)
      unlinkFiles(done)
    } else {
      log.info('done')
      done()
    }
  })

  function unless (flag, step) {
    return function (done) {
      if (flag) {
        done()
      } else {
        step(done)
      }
    }
  }

  function checkHaveRecord (done) {
    var file = recordPath(configuration.directory, publication.digest)
    fileExists(file, function (error, exists) {
      if (error) return done(error)
      haveRecord = exists
      log.info({haveRecord: exists})
      done()
    })
  }

  function checkHavePeerTimestamp (done) {
    var file = timestampPath(
      configuration.directory,
      publication.digest,
      peer.publicKey
    )
    fileExists(file, function (error, exists) {
      if (error) return done(error)
      havePeerTimestamp = exists
      log.info({havePeerTimestamp: exists})
      done()
    })
  }

  function getRecord (done) {
    if (haveRecord) {
      readRecord(
        configuration.directory, publication.digest,
        function (error, parsed) {
          if (error) return done(error)
          record = parsed
          done()
        }
      )
    } else {
      http.request(xtend(peer.url, {
        path: peer.url.path + '/publications/' + publication.digest,
        headers: {accept: 'application/json'}
      }))
        .once('error', function (error) {
          done(error)
        })
        .once('response', function (response) {
          pump(
            response,
            concat(function (buffer) {
              parse(buffer, function (error, parsed) {
                if (error) return done(error)
                record = parsed
                log.info('got record')
                done()
              })
            }),
            function (error) {
              if (error) {
                done(error)
              }
            }
          )
        })
        .end()
    }
  }

  function getPeerTimestamp (done) {
    var request = xtend(peer.url, {
      path: (
        peer.url.path +
        'publications/' + publication.digest +
        '/timestamps/' + encoding.encode(peer.publicKey)
      )
    })
    http.request(request)
      .once('error', function (error) {
        done(error)
      })
      .once('response', function (response) {
        if (response.statusCode !== 200) {
          var error = new Error('Could not get timestamp')
          error.statusCode = response.statusCode
          log.error({
            statusCode: response.statusCode
          }, 'failed to get timestamp')
          done(error)
        } else {
          pump(
            response,
            concat(function (buffer) {
              parse(buffer, function (error, parsed) {
                if (error) return done(error)
                peerTimestamp = parsed
                log.info('got timestamp')
                done()
              })
            }),
            function (error) {
              if (error) {
                done(error)
              }
            }
          )
        }
      })
      .end()
  }

  function validateRecord (done) {
    validatePublication(record)
    var errors = validatePublication.errors
    if (errors) {
      log.info({errors: errors}, 'invalid publication')
      var error = new Error('invalid record')
      error.validationErrors = errors
      return done(error)
    }
    var computedDigest = encoding.encode(
      sodium.crypto_hash_sha256(
        Buffer.from(stringify(record), 'utf8')
      )
    )
    if (computedDigest !== publication.digest) {
      done(new Error(
        'reported and computed digests do not match'
      ))
    }
    log.info('validated record')
    done()
  }

  function validatePeerTimestamp (done) {
    validateTimestamp(peerTimestamp)
    var errors = validateTimestamp.errors
    if (errors) {
      log.info({errors: errors}, 'invalid timestamp')
      var error = new Error('invalid timestamp')
      error.validationErrors = errors
      return done(error)
    }
    var digestsMatch = (
      peerTimestamp.timestamp.digest ===
      encoding.encode(
        sodium.crypto_hash_sha256(
          Buffer.from(stringify(record), 'utf8')
        )
      )
    )
    if (!digestsMatch) {
      log.error('timestamp digest mismatch')
      return done(new Error(
        'peer timestamp digest does not match record'
      ))
    }
    var validSignature = sodium.crypto_sign_verify_detached(
      encoding.decode(peerTimestamp.signature),
      Buffer.from(stringify(peerTimestamp.timestamp)),
      peer.publicKey
    )
    if (!validSignature) {
      log.error('invalid peer signature')
      return done(new Error('invalid peer signature'))
    }
    log.info('validated timestamp')
    done()
  }

  function makeDirectory (done) {
    mkdirp(
      recordDirectoryPath(configuration.directory, publication.digest),
      done
    )
  }

  function saveOwnTimestamp (done) {
    var timestamp = {
      digest: publication,
      uri: (
        'https://' + configuration.hostname +
        '/publications/' + publication.digest
      ),
      time: time
    }
    var signature = encoding.encode(
      sodium.crypto_sign_detached(
        Buffer.from(stringify(timestamp)),
        configuration.keypair.secret
      )
    )
    var file = signaturePath(
      configuration.directory,
      publication.digest,
      encoding.encode(configuration.keypair.public)
    )
    files.push(file)
    var data = stringify({
      timestamp: timestamp,
      signature: signature,
      version: timestampSchema.properties.version.constant
    })
    fs.writeFile(file, data, 'utf8', function (error) {
      if (error) return done(error)
      log.info('saved own timestamp')
      done()
    })
  }

  function savePeerTimestamp (done) {
    var file = timestampPath(
      configuration.directory,
      publication.digest,
      peer.publicKey
    )
    files.push(file)
    var data = stringify(peerTimestamp)
    fs.writeFile(file, data, 'utf8', function (error) {
      if (error) return done(error)
      log.info('saved peer timestamp')
      done()
    })
  }

  function saveRecord (done) {
    var file = recordPath(configuration.directory, publication.digest)
    files.push(file)
    var data = stringify(record)
    writeIfMissing(file, data, function (error) {
      if (error) return done(error)
      log.info('saved record')
      done()
    })
  }

  function getAndSaveAttachments (done) {
    runParallel(
      record.attachments.map(function (attachmentDigest) {
        return function (done) {
          var attachmentLog = log.child({
            attachment: attachmentDigest
          })
          var request = xtend(peer.url, {
            path: (
              peer.url.path +
              '/publications/' + publication.digest +
              '/attachments/' + attachmentDigest
            )
          })
          var file = attachmentPath(
            configuration.directory,
            publication.digest,
            attachmentDigest
          )
          download(request, file, function (error, contentType) {
            if (error) return done(error)
            attachmentLog.info('downloaded')
            writeIfMissing(
              file + '.type', contentType,
              function (error) {
                if (error) return done(error)
                attachmentLog.info('wrote type file')
                done()
              }
            )
          })
        }
      }),
      done
    )
  }

  function appendToAccessions (done) {
    fs.appendFile(
      path.join(configuration.directory, 'accessions'),
      (time || new Date().toISOString()) + ',' +
      publication.digest + '\n',
      done
    )
  }

  function bumpAccessionNumber (done) {
    peer.last = publication.accessionNumber
    log.info('bumped')
    done()
  }

  function unlinkFiles (done) {
    runParallel(files.map(function (file) {
      return function (done) {
        var fileLog = log.child({file: file})
        fs.unlink(file, function (error) {
          if (error) {
            fileLog.error(error)
            done()
          } else {
            fileLog.info('unlinked')
            done()
          }
        })
      }
    }), done)
  }
}

function writeIfMissing (path, data, done) {
  fs.writeFile(path, data, {flag: 'wx'}, function (error) {
    if (error && error.code !== 'EEXIST') {
      done(error)
    } else {
      done()
    }
  })
}

function download (from, to, done) {
  var contentType
  http.request(from)
    .once('error', function (error) {
      done(error)
    })
    .once('response', function (response) {
      contentType = response.headers['Content-Type']
      pump(
        response,
        fs.createWriteStream(to, {flags: 'wx'}),
        function (error) {
          if (error) {
            if (error.code === 'EEXIST') {
              done()
            } else {
              done(error)
            }
          } else {
            done(null, contentType)
          }
        }
      )
    })
    .end()
}

function readPeers (directory, log, callback) {
  var file = path.join(directory, 'peers')
  fs.readFile(file, 'utf8', function (error, string) {
    if (error) {
      if (error.code === 'ENOENT') {
        callback(null, [])
      } else {
        callback(error)
      }
    } else {
      var peers = string
        .split('\n')
        .reduce(function (peers, line, index) {
          var split = line.split(',')
          if (split.length === 3) {
            try {
              var peer = {
                url: url.parse(split[0]),
                publicKey: encoding.decode(split[1]),
                last: parseInt(split[2])
              }
              peers.push(peer)
            } catch (error) {
              log.error('peers parse error', {
                line: line,
                number: index + 1
              })
            }
          }
          return peers
        }, [])
      callback(null, peers)
    }
  })
}

function writePeers (directory, peers, callback) {
  var file = path.join(directory, 'peers')
  var data = peers
    .map(function (peer) {
      return (
        url.format(peer.url) + ',' +
        encoding.encode(peer.publicKey) + ',' +
        peer.last.toString()
      )
    })
    .join('\n')
  fs.writeFile(file, data, callback)
}

function fileExists (path, callback) {
  fs.access(path, fs.constants.R_OK, function (error) {
    if (error) {
      if (error.code === 'ENOENT') {
        callback(null, false)
      } else {
        callback(error)
      }
    } else {
      callback(null, true)
    }
  })
}