Rework crawler progress events

Author: pfrazee

crawler/followgraph.js

@@ -6,7 +6,7 @@ const lock = require('../lib/lock')
 const db = require('../dbs/profile-data-db')
 const crawler = require('./index')
 const siteDescriptions = require('./site-descriptions')
-const {doCrawl, doCheckpoint} = require('./util')
+const {doCrawl, doCheckpoint, emitProgressEvent} = require('./util')
 const debug = require('../lib/debug-logger').debugLogger('crawler')
 
 // constants
@@ -46,6 +46,8 @@ exports.crawlSite = async function (archive, crawlSource) {
       return
     }
 
+    emitProgressEvent(archive.url, 'crawl_followgraph', 0, 1)
+
     // read and validate
     try {
       var followsJson = await readFollowsFile(archive)
@@ -91,6 +93,7 @@ exports.crawlSite = async function (archive, crawlSource) {
 
     // write checkpoint as success
     await doCheckpoint('crawl_followgraph', TABLE_VERSION, crawlSource, changes[changes.length - 1].version)
+    emitProgressEvent(archive.url, 'crawl_followgraph', 1, 1)
   })
 }
 

crawler/posts.js

@@ -3,7 +3,7 @@ const {URL} = require('url')
 const Events = require('events')
 const db = require('../dbs/profile-data-db')
 const crawler = require('./index')
-const {doCrawl, doCheckpoint, getMatchingChangesInOrder, generateTimeFilename} = require('./util')
+const {doCrawl, doCheckpoint, emitProgressEvent, getMatchingChangesInOrder, generateTimeFilename} = require('./util')
 const debug = require('../lib/debug-logger').debugLogger('crawler')
 
 // constants
@@ -40,8 +40,10 @@ exports.crawlSite = async function (archive, crawlSource) {
     // collect changed posts
     var changedPosts = getMatchingChangesInOrder(changes, JSON_PATH_REGEX)
     console.log('collected changed posts', changedPosts)
+    emitProgressEvent(archive.url, 'crawl_posts', 0, changedPosts.length)
 
     // read and apply each post in order
+    var progress = 0
     for (let changedPost of changedPosts) {
       // TODO Currently the crawler will abort reading the feed if any post fails to load
       //      this means that a single bad or unreachable file can stop the forward progress of post indexing
@@ -89,10 +91,11 @@ exports.crawlSite = async function (archive, crawlSource) {
           `, [crawlSource.id, changedPost.name, Date.now(), post.content, post.createdAt, post.updatedAt])
           events.emit('post-added', archive.url)
         }
-
-        // checkpoint our progress
-        await doCheckpoint('crawl_posts', TABLE_VERSION, crawlSource, changedPost.version)
       }
+
+      // checkpoint our progress
+      await doCheckpoint('crawl_posts', TABLE_VERSION, crawlSource, changedPost.version)
+      emitProgressEvent(archive.url, 'crawl_posts', ++progress, changedPosts.length)
     }
   })
 }

crawler/site-descriptions.js

@@ -6,7 +6,7 @@ const db = require('../dbs/profile-data-db')
 const archivesDb = require('../dbs/archives')
 const dat = require('../dat')
 const crawler = require('./index')
-const {doCrawl, doCheckpoint, getMatchingChangesInOrder, generateTimeFilename} = require('./util')
+const {doCrawl, doCheckpoint, emitProgressEvent, getMatchingChangesInOrder, generateTimeFilename} = require('./util')
 const debug = require('../lib/debug-logger').debugLogger('crawler')
 
 // constants
@@ -43,8 +43,10 @@ exports.crawlSite = async function (archive, crawlSource) {
     // collect changed site descriptions
     var changedSiteDescriptions = getMatchingChangesInOrder(changes, JSON_PATH_REGEX)
     console.log('collected changed site descriptions', changedSiteDescriptions)
+    emitProgressEvent(archive.url, 'crawl_site_descriptions', 0, changedSiteDescriptions.length)
 
     // read and apply each post in order
+    var progress = 0
     for (let changedSiteDescription of changedSiteDescriptions) {
       // TODO Currently the crawler will abort reading the feed if any description fails to load
       //      this means that a single bad or unreachable file can stop the forward progress of description indexing
@@ -95,10 +97,11 @@ exports.crawlSite = async function (archive, crawlSource) {
             VALUES (?, ?, ?, ?, ?, ?, ?, ?)
         `, [crawlSource.id, changedSiteDescription.name, Date.now(), desc.subject, desc.metadata.title, desc.metadata.description, desc.metadata.type.join(','), desc.createdAt])
         events.emit('description-added', archive.url)
-
-        // checkpoint our progress
-        await doCheckpoint('crawl_site_descriptions', TABLE_VERSION, crawlSource, changedSiteDescription.version)
       }
+
+      // checkpoint our progress
+      await doCheckpoint('crawl_site_descriptions', TABLE_VERSION, crawlSource, changedSiteDescription.version)
+      emitProgressEvent(archive.url, 'crawl_site_descriptions', ++progress, changedSiteDescription.length)
     }
   })
 }

crawler/util.js

@@ -54,7 +54,7 @@ exports.doCrawl = async function (archive, crawlSource, crawlDataset, crawlDatas
 }
 
 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
@@ -63,6 +63,10 @@ const doCheckpoint = exports.doCheckpoint = async function (crawlDataset, crawlD
   `, [crawlDataset, crawlDatasetVersion, crawlSource.id, crawlSourceVersion, Date.now()])
 }
 
+exports.emitProgressEvent = function (sourceUrl, crawlDataset, progress, numUpdates) {
+  crawlerEvents.emit('crawl-dataset-progress', {sourceUrl, crawlDataset, progress, numUpdates})
+}
+
 exports.getMatchingChangesInOrder = function (changes, regex) {
   var list = [] // order matters, must be oldest to newest
   changes.forEach(c => {