Add beaker.crawler API

Author: pfrazee

crawler/index.js

@@ -1,9 +1,13 @@
+const emitStream = require('emit-stream')
+const {URL} = require('url')
 const _throttle = require('lodash.throttle')
 const lock = require('../lib/lock')
 const db = require('../dbs/profile-data-db')
+const archivesDb = require('../dbs/archives')
 const users = require('../users')
 const dat = require('../dat')
 
+const {crawlerEvents} = require('./util')
 const posts = require('./posts')
 const followgraph = require('./followgraph')
 const siteDescriptions = require('./site-descriptions')
@@ -21,6 +25,7 @@ const watches = {}
 exports.posts = posts
 exports.followgraph = followgraph
 exports.siteDescriptions = siteDescriptions
+const createEventsStream = exports.createEventsStream = () => emitStream(crawlerEvents)
 
 exports.setup = async function () {
 }
@@ -32,6 +37,7 @@ exports.watchSite = async function (archive) {
   console.log('watchSite', archive.url)
 
   if (!(archive.url in watches)) {
+    crawlerEvents.emit('watch', {sourceUrl: archive.url})
     const queueCrawl = _throttle(() => crawlSite(archive), 5e3)
 
     // watch for file changes
@@ -57,6 +63,7 @@ exports.watchSite = async function (archive) {
 exports.unwatchSite = async function (url) {
   // stop watching for file changes
   if (url in watches) {
+    crawlerEvents.emit('unwatch', {sourceUrl: url})
     watches[url].close()
     watches[url] = null
   }
@@ -65,6 +72,7 @@ exports.unwatchSite = async function (url) {
 const crawlSite =
 exports.crawlSite = async function (archive) {
   console.log('crawling', archive.url)
+  crawlerEvents.emit('crawl-start', {sourceUrl: archive.url})
   var release = await lock('crawl:' + archive.url)
   try {
     // get/create crawl source
@@ -80,7 +88,49 @@ exports.crawlSite = async function (archive) {
       followgraph.crawlSite(archive, crawlSource),
       siteDescriptions.crawlSite(archive, crawlSource)
     ])
+  } catch (err) {
+    crawlerEvents.emit('crawl-error', {sourceUrl: archive.url, err: err.toString()})
   } finally {
+    crawlerEvents.emit('crawl-finish', {sourceUrl: archive.url})
     release()
   }
-}
+}
+
+const getCrawlStates =
+exports.getCrawlStates = async function () {
+  var rows = await db.all(`
+    SELECT
+        crawl_sources.url AS url,
+        GROUP_CONCAT(crawl_sources_meta.crawlSourceVersion) AS versions,
+        GROUP_CONCAT(crawl_sources_meta.crawlDataset) AS datasets,
+        MAX(crawl_sources_meta.updatedAt) AS updatedAt
+      FROM crawl_sources
+      INNER JOIN crawl_sources_meta ON crawl_sources_meta.crawlSourceId = crawl_sources.id
+      GROUP BY crawl_sources.id
+  `)
+  return Promise.all(rows.map(async ({url, versions, datasets, updatedAt}) => {
+    var datasetVersions = {}
+    versions = versions.split(',')
+    datasets = datasets.split(',')
+    for (let i = 0; i < datasets.length; i++) {
+      datasetVersions[datasets[i]] = Number(versions[i])
+    }
+    var meta = await archivesDb.getMeta(toHostname(url))
+    return {url, title: meta.title, datasetVersions, updatedAt}
+  }))
+}
+
+const resetSite =
+exports.resetSite = async function (url) {
+  await db.run(`DELETE FROM crawl_sources WHERE url = ?`, [url])
+}
+
+exports.WEBAPI = {createEventsStream, getCrawlStates, resetSite}
+
+// internal methods
+// =
+
+function toHostname (url) {
+  url = new URL(url)
+  return url.hostname
+}

crawler/util.js

@@ -1,10 +1,14 @@
+const EventEmitter = require('events')
 const pump = require('pump')
 const concat = require('concat-stream')
 const db = require('../dbs/profile-data-db')
 const dat = require('../dat')
 
 const READ_TIMEOUT = 30e3
 
+const crawlerEvents = new EventEmitter()
+exports.crawlerEvents = crawlerEvents
+
 exports.doCrawl = async function (archive, crawlSource, crawlDataset, crawlDatasetVersion, handlerFn) {
   const url = archive.url
 
@@ -38,14 +42,19 @@ exports.doCrawl = async function (archive, crawlSource, crawlDataset, crawlDatas
     )
   })
 
+  crawlerEvents.emit('crawl-dataset-start', {sourceUrl: archive.url, crawlDataset, crawlRange: {start, end}})
+
   // handle changes
   await handlerFn({changes, resetRequired})
 
   // final checkpoint
   await doCheckpoint(crawlDataset, crawlDatasetVersion, crawlSource, version)
+
+  crawlerEvents.emit('crawl-dataset-finish', {sourceUrl: archive.url, crawlDataset, crawlRange: {start, end}})
 }
 
 const doCheckpoint = exports.doCheckpoint = async function (crawlDataset, crawlDatasetVersion, crawlSource, crawlSourceVersion) {
+  crawlerEvents.emit('crawl-dataset-progress', {sourceUrl: crawlSource.url, crawlDataset, crawledVersion: crawlSourceVersion})
   await db.run(`DELETE FROM crawl_sources_meta WHERE crawlDataset = ? AND crawlSourceId = ?`, [crawlDataset, crawlSource.id])
   await db.run(`
     INSERT

web-apis/bg.js

@@ -10,6 +10,7 @@ const downloadsManifest = require('./manifests/internal/downloads')
 const historyManifest = require('./manifests/internal/history')
 const sitedataManifest = require('./manifests/internal/sitedata')
 const watchlistManifest = require('./manifests/internal/watchlist')
+const crawlerManifest = require('./manifests/internal/crawler')
 const postsManifest = require('./manifests/internal/posts')
 const followgraphManifest = require('./manifests/internal/followgraph')
 
@@ -19,6 +20,7 @@ const bookmarksAPI = require('./bg/bookmarks')
 const historyAPI = require('./bg/history')
 const sitedataAPI = require('../dbs/sitedata').WEBAPI
 const watchlistAPI = require('./bg/watchlist')
+const crawlerAPI = require('../crawler').WEBAPI
 const postsAPI = require('./bg/posts')
 const followgraphAPI = require('./bg/followgraph')
 
@@ -54,6 +56,7 @@ exports.setup = function () {
   globals.rpcAPI.exportAPI('history', historyManifest, historyAPI, internalOnly)
   globals.rpcAPI.exportAPI('sitedata', sitedataManifest, sitedataAPI, internalOnly)
   globals.rpcAPI.exportAPI('watchlist', watchlistManifest, watchlistAPI, internalOnly)
+  globals.rpcAPI.exportAPI('crawler', crawlerManifest, crawlerAPI, internalOnly)
   globals.rpcAPI.exportAPI('posts', postsManifest, postsAPI, internalOnly)
   globals.rpcAPI.exportAPI('followgraph', followgraphManifest, followgraphAPI, internalOnly)
 

web-apis/fg/beaker.js

@@ -8,6 +8,7 @@ const downloadsManifest = require('../manifests/internal/downloads')
 const historyManifest = require('../manifests/internal/history')
 const sitedataManifest = require('../manifests/internal/sitedata')
 const watchlistManifest = require('../manifests/internal/watchlist')
+const crawlerManifest = require('../manifests/internal/crawler')
 const postsManifest = require('../manifests/internal/posts')
 const followgraphManifest = require('../manifests/internal/followgraph')
 
@@ -24,6 +25,7 @@ exports.setup = function (rpc) {
     const historyRPC = rpc.importAPI('history', historyManifest, opts)
     const sitedataRPC = rpc.importAPI('sitedata', sitedataManifest, opts)
     const watchlistRPC = rpc.importAPI('watchlist', watchlistManifest, opts)
+    const crawlerRPC = rpc.importAPI('crawler', crawlerManifest, opts)
     const postsRPC = rpc.importAPI('posts', postsManifest, opts)
     const followgraphRPC = rpc.importAPI('followgraph', followgraphManifest, opts)
 
@@ -158,6 +160,12 @@ exports.setup = function (rpc) {
     beaker.watchlist.remove = watchlistRPC.remove
     beaker.watchlist.createEventsStream = () => fromEventStream(watchlistRPC.createEventsStream())
 
+    // beaker.crawler
+    beaker.crawler = {}
+    beaker.crawler.getCrawlStates = crawlerRPC.getCrawlStates
+    beaker.crawler.resetSite = crawlerRPC.resetSite
+    beaker.crawler.createEventsStream = () => fromEventStream(crawlerRPC.createEventsStream())
+
     // beaker.posts
     beaker.posts = {}
     beaker.posts.list = postsRPC.list

web-apis/manifests/internal/crawler.js

@@ -0,0 +1,5 @@
+module.exports = {
+  getCrawlStates: 'promise',
+  resetSite: 'promise',
+  createEventsStream: 'readable'
+}