db migration 1.0.2, now stores and indexes date of content delivered to subscriber...

[websub-hub] / src / communication.js

diff --git a/src/communication.js b/src/communication.js
index 72f2642a003e465aab42909ce13f90b8fd93694c..dc4d464c70111b4ea2a8971575940c099d14b379 100644 (file)
--- a/src/communication.js
+++ b/src/communication.js
@@ -49,7 +49,7 @@ class Communication {
       return response;
     });
 
-    this.worker = new Worker(logger, this.workFeed.bind(this), options);
+    this.worker = new Worker(logger, db, this.workFeed.bind(this), options);
     this.worker.start();
   }
 
@@ -57,7 +57,7 @@ class Communication {
   static userAgentString(userAgentConfig) {
     // eslint-disable-next-line security/detect-object-injection
     const _conf = (field, def) => (userAgentConfig && field in userAgentConfig) ? userAgentConfig[field] : def;
-    const product = _conf('product', packageName);
+    const product = _conf('product', packageName).split('/').pop();
     const version = _conf('version', packageVersion);
     let implementation = _conf('implementation', Enum.Specification);
     if (implementation) {
@@ -223,6 +223,7 @@ class Communication {
     }
 
     if (!topic.isActive) {
+      // These should be filtered out when selecting verification tasks to process.
       this.logger.debug(_scope, 'topic not active, skipping verification', { verification, requestId });
       await this.db.verificationRelease(dbCtx, verificationId);
       return;
@@ -328,11 +329,19 @@ class Communication {
         case Enum.Mode.Unsubscribe:
           if (verificationAccepted) {
             await this.db.subscriptionDelete(txCtx, verification.callback, verification.topicId);
+            if (topic.isDeleted) {
+              // Remove a deleted topic after the last subscription is notified.
+              await this.db.topicPendingDelete(txCtx, topic.id);
+            }
           }
           break;
 
         case Enum.Mode.Denied:
           await this.db.subscriptionDelete(txCtx, verification.callback, verification.topicId);
+          if (topic.isDeleted) {
+            // Remove a deleted topic after he last subscription is notified.
+            await this.db.topicPendingDelete(txCtx, topic.id);
+          }
           break;
 
         default:
@@ -340,7 +349,7 @@ class Communication {
           throw new Errors.InternalInconsistencyError(verification.mode);
       }
 
-      await this.db.verificationComplete(dbCtx, verificationId, verification.callback, verification.topicId);
+      await this.db.verificationComplete(txCtx, verificationId, verification.callback, verification.topicId);
     }); // txCtx
 
     this.logger.info(_scope, 'verification complete', { ...logInfoData, verificationAccepted });
@@ -431,7 +440,10 @@ class Communication {
       throw new Errors.InternalInconsistencyError('no such topic id');
     }
 
-    logInfoData.url = topicId.url;
+    // Cull any expired subscriptions
+    await this.db.subscriptionDeleteExpired(dbCtx, topicId);
+
+    logInfoData.url = topic.url;
 
     if (topic.isDeleted) {
       this.logger.debug(_scope, 'topic deleted, skipping update request', logInfoData);
@@ -479,13 +491,15 @@ class Communication {
 
     const validHub = await this.linkHelper.validHub(topic.url, response.headers, response.data);
     if (!validHub) {
-      this.logger.debug(_scope, 'retrieved topic does not list us as hub', { logInfoData });
+      this.logger.info(_scope, 'retrieved topic does not list us as hub', { logInfoData });
       if (this.options.communication.strictTopicHubLink) {
         await this.db.transaction(dbCtx, async (txCtx) => {
           // Set as deleted and set content_updated so subscriptions are notified.
           await this.db.topicDeleted(txCtx, topicId);
           await this.db.topicFetchComplete(txCtx, topicId);
         });
+        // Attempt to remove from db, if no active subscriptions.
+        await this.db.topicPendingDelete(dbCtx, topicId);
         return;
       }
     }
@@ -551,7 +565,7 @@ class Communication {
 
       await this.db.transaction(dbCtx, async (txCtx) => {
         await this.db.verificationInsert(txCtx, verification);
-        await this.db.subscriptionDeliveryComplete(txCtx, subscription.callback, subscription.topicId);
+        await this.db.subscriptionDeliveryComplete(txCtx, subscription.callback, subscription.topicId, topic.contentUpdated);
       });
       this.logger.info(_scope, 'update unsubscription for deleted topic', logInfoData);
       return;
@@ -603,7 +617,7 @@ class Communication {
         return;
     }
 
-    await this.db.subscriptionDeliveryComplete(dbCtx, subscription.callback, subscription.topicId);
+    await this.db.subscriptionDeliveryComplete(dbCtx, subscription.callback, subscription.topicId, topic.contentUpdated);
     this.logger.info(_scope, 'update success', logInfoData);
   }
 
@@ -647,10 +661,11 @@ class Communication {
 
   /**
    * 
+   * @param {*} dbCtx
    * @param {Number} wanted maximum tasks to claim
    * @returns {Promise<void>[]}
    */
-  async workFeed(wanted) {
+  async workFeed(dbCtx, wanted) {
     const _scope = _fileScope('workFeed');
     const inProgress = [];
     const requestId = common.requestId();
@@ -661,33 +676,29 @@ class Communication {
     this.logger.debug(_scope, 'called', { wanted });
 
     try {
-      await this.db.context(async (dbCtx) => {
-        if (wanted > 0) {
-          // Update topics before anything else.
-          const topicFetchIds = await this.db.topicFetchClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
-          // Each task gets a new context, as these map to connections in some dbs.
-          // This dbCtx goes away after launching the processing functions, so would not be available to tasks.
-          topicFetchPromises = topicFetchIds.map((id) => this.db.context((ctx) => this.topicFetchProcess(ctx, id, requestId)));
-          inProgress.push(...topicFetchPromises);
-          wanted -= topicFetchPromises.length;
-        }
+      if (wanted > 0) {
+        // Update topics before anything else.
+        const topicFetchIds = await this.db.topicFetchClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
+        topicFetchPromises = topicFetchIds.map((id) => this.db.context((ctx) => this.topicFetchProcess(ctx, id, requestId)));
+        inProgress.push(...topicFetchPromises);
+        wanted -= topicFetchPromises.length;
+      }
 
-        if (wanted > 0) {
-          // Then any pending verifications.
-          const verifications = await this.db.verificationClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
-          verificationPromises = verifications.map((id) => this.db.context((ctx) => this.verificationProcess(ctx, id, requestId)));
-          inProgress.push(...verificationPromises);
-          wanted -= verificationPromises.length;
-        }
+      if (wanted > 0) {
+        // Then any pending verifications.
+        const verifications = await this.db.verificationClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
+        verificationPromises = verifications.map((id) => this.db.context((ctx) => this.verificationProcess(ctx, id, requestId)));
+        inProgress.push(...verificationPromises);
+        wanted -= verificationPromises.length;
+      }
 
-        if (wanted > 0) {
-          // Finally dole out content.
-          const updates = await this.db.subscriptionDeliveryClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
-          updatePromises = updates.map((id) => this.db.context((ctx) => this.subscriptionDeliveryProcess(ctx, id, requestId)));
-          inProgress.push(...updatePromises);
-          wanted -= updatePromises.length;
-        }
-      }); // dbCtx
+      if (wanted > 0) {
+        // Finally dole out content.
+        const updates = await this.db.subscriptionDeliveryClaim(dbCtx, wanted, claimTimeoutSeconds, nodeId);
+        updatePromises = updates.map((id) => this.db.context((ctx) => this.subscriptionDeliveryProcess(ctx, id, requestId)));
+        inProgress.push(...updatePromises);
+        wanted -= updatePromises.length;
+      }
     } catch (e) {
       this.logger.error(_scope, 'failed', { error: e });
       // do not re-throw, return what we've claimed so far