Merge pull request #1242 from qtomlinson/qt/definition_upgrade

Add the ability to queue definition recomputation

Author: qtomlinson

app.js

@@ -59,6 +59,9 @@ function createApp(config) {
   const harvestQueue = config.harvest.queue()
   initializers.push(async () => harvestQueue.initialize())
 
+  const upgradeHandler = config.upgrade.service({ queue: config.upgrade.queue })
+  initializers.push(async () => upgradeHandler.initialize())
+
   const definitionService = require('./business/definitionService')(
     harvestStore,
     harvestService,
@@ -67,7 +70,8 @@ function createApp(config) {
     curationService,
     definitionStore,
     searchService,
-    cachingService
+    cachingService,
+    upgradeHandler
   )
   // Circular dependency. Reach in and set the curationService's definitionService. Sigh.
   curationService.definitionService = definitionService
@@ -234,6 +238,7 @@ function createApp(config) {
         // kick off the queue processors
         require('./providers/curation/process')(curationQueue, curationService, logger)
         require('./providers/harvest/process')(harvestQueue, definitionService, logger)
+        upgradeHandler.setupProcessing(definitionService, logger)
 
         // Signal system is up and ok (no error)
         callback()

bin/config.js

@@ -57,6 +57,10 @@ module.exports = {
   definition: {
     store: loadFactory(config.get('DEFINITION_STORE_PROVIDER') || 'file', 'definition')
   },
+  upgrade: {
+    queue: loadFactory(config.get('DEFINITION_UPGRADE_QUEUE_PROVIDER') || 'memory', 'upgrade.queue'),
+    service: loadFactory(config.get('DEFINITION_UPGRADE_PROVIDER') || 'versionCheck', 'upgrade.service')
+  },
   attachment: {
     store: loadFactory(config.get('ATTACHMENT_STORE_PROVIDER') || 'file', 'attachment')
   },

business/definitionService.js

@@ -39,7 +39,7 @@ const currentSchema = '1.7.0'
 const weights = { declared: 30, discovered: 25, consistency: 15, spdx: 15, texts: 15, date: 30, source: 70 }
 
 class DefinitionService {
-  constructor(harvestStore, harvestService, summary, aggregator, curation, store, search, cache) {
+  constructor(harvestStore, harvestService, summary, aggregator, curation, store, search, cache, upgradeHandler) {
     this.harvestStore = harvestStore
     this.harvestService = harvestService
     this.summaryService = summary
@@ -48,9 +48,15 @@ class DefinitionService {
     this.definitionStore = store
     this.search = search
     this.cache = cache
+    this.upgradeHandler = upgradeHandler
+    if (this.upgradeHandler) this.upgradeHandler.currentSchema = currentSchema
     this.logger = logger()
   }
 
+  get currentSchema() {
+    return currentSchema
+  }
+
   /**
    * Get the final representation of the specified definition and optionally apply the indicated
    * curation.
@@ -68,11 +74,10 @@ class DefinitionService {
       return this.compute(coordinates, curation)
     }
     const existing = await this._cacheExistingAside(coordinates, force)
-    let result
-    if (get(existing, '_meta.schemaVersion') === currentSchema) {
+    let result = await this.upgradeHandler.validate(existing)
+    if (result) {
       // Log line used for /status page insights
       this.logger.info('computed definition available', { coordinates: coordinates.toString() })
-      result = existing
     } else result = force ? await this.computeAndStore(coordinates) : await this.computeStoreAndCurate(coordinates)
     return this._trimDefinition(this._cast(result), expand)
   }
@@ -598,5 +603,15 @@ class DefinitionService {
   }
 }
 
-module.exports = (harvestStore, harvestService, summary, aggregator, curation, store, search, cache) =>
-  new DefinitionService(harvestStore, harvestService, summary, aggregator, curation, store, search, cache)
+module.exports = (harvestStore, harvestService, summary, aggregator, curation, store, search, cache, versionHandler) =>
+  new DefinitionService(
+    harvestStore,
+    harvestService,
+    summary,
+    aggregator,
+    curation,
+    store,
+    search,
+    cache,
+    versionHandler
+  )

providers/index.js

@@ -27,6 +27,16 @@ module.exports = {
   auth: {
     github: require('../middleware/githubConfig')
   },
+  upgrade: {
+    queue: {
+      azure: require('../providers/upgrade/azureQueueConfig'),
+      memory: require('../providers/upgrade/memoryQueueConfig')
+    },
+    service: {
+      versionCheck: require('../providers/upgrade/defVersionCheck').factory,
+      upgradeQueue: require('../providers/upgrade/defUpgradeQueueConfig')
+    }
+  },
   curation: {
     queue: {
       azure: require('../providers/curation/azureQueueConfig'),

providers/queueing/memoryQueue.js

@@ -9,6 +9,7 @@ class MemoryQueue {
     this.logger = logger()
     this.data = []
     this.messageId = 0
+    this.decoder = options.decoder
   }
 
   async initialize() {}
@@ -33,14 +34,19 @@ class MemoryQueue {
     const message = this.data[0]
     if (!message) return null
     this.data[0].dequeueCount++
-    if (message.dequeueCount <= 5) return Promise.resolve({ original: message, data: JSON.parse(message.messageText) })
+    if (message.dequeueCount <= 5) return Promise.resolve({ original: message, data: this._parseData(message) })
     await this.delete({ original: message })
     return this.dequeue()
   }
 
+  _parseData({ messageText }) {
+    return JSON.parse(this.decoder(messageText))
+  }
+
   /** Similar to dequeue() but returns an array instead. See AzureStorageQueue.dequeueMultiple() */
   async dequeueMultiple() {
-    return [await this.dequeue()]
+    const message = await this.dequeue()
+    return message ? [message] : []
   }
 
   /**
@@ -58,4 +64,12 @@ class MemoryQueue {
   }
 }
 
-module.exports = () => new MemoryQueue()
+const factory = (opts = {}) => {
+  const defaultOpts = {
+    decoder: text => text
+  }
+  const mergedOpts = { ...defaultOpts, ...opts }
+  return new MemoryQueue(mergedOpts)
+}
+
+module.exports = factory

providers/upgrade/azureQueueConfig.js

@@ -0,0 +1,22 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const config = require('painless-config')
+const AzureStorageQueue = require('../queueing/azureStorageQueue')
+
+const defaultOptions = {
+  connectionString:
+    config.get('DEFINITION_UPGRADE_QUEUE_CONNECTION_STRING') || config.get('HARVEST_AZBLOB_CONNECTION_STRING'),
+  queueName: config.get('DEFINITION_UPGRADE_QUEUE_NAME') || 'definitions-upgrade',
+  dequeueOptions: {
+    numOfMessages: 32,
+    visibilityTimeout: 10 * 60 // 10 min. The default value is 30 seconds.
+  }
+}
+
+function azure(options) {
+  const realOptions = options || defaultOptions
+  return new AzureStorageQueue(realOptions)
+}
+
+module.exports = azure

providers/upgrade/defUpgradeQueue.js

@@ -0,0 +1,50 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const { DefinitionVersionChecker } = require('./defVersionCheck')
+const { setup } = require('./process')
+
+class DefinitionQueueUpgrader extends DefinitionVersionChecker {
+  async validate(definition) {
+    if (!definition) return
+    const result = await super.validate(definition)
+    if (result) return result
+
+    await this._queueUpgrade(definition)
+    return definition
+  }
+
+  async _queueUpgrade(definition) {
+    if (!this._upgrade) throw new Error('Upgrade queue is not set')
+    try {
+      const message = this._constructMessage(definition)
+      await this._upgrade.queue(message)
+      this.logger.debug('Queued for definition upgrade ', {
+        coordinates: DefinitionVersionChecker.getCoordinates(definition)
+      })
+    } catch (error) {
+      //continue if queuing fails and requeue at the next request.
+      this.logger.error(`Error queuing for definition upgrade ${error.message}`, {
+        error,
+        coordinates: DefinitionVersionChecker.getCoordinates(definition)
+      })
+    }
+  }
+
+  _constructMessage(definition) {
+    const { coordinates, _meta } = definition
+    const content = { coordinates, _meta }
+    return Buffer.from(JSON.stringify(content)).toString('base64')
+  }
+
+  async initialize() {
+    this._upgrade = this.options.queue()
+    return this._upgrade.initialize()
+  }
+
+  setupProcessing(definitionService, logger, once) {
+    return setup(this._upgrade, definitionService, logger, once)
+  }
+}
+
+module.exports = DefinitionQueueUpgrader

providers/upgrade/defUpgradeQueueConfig.js

@@ -0,0 +1,12 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const DefinitionQueueUpgrader = require('./defUpgradeQueue')
+const memory = require('../queueing/memoryQueue')
+
+function serviceFactory(options = {}) {
+  const mergedOptions = { queue: memory, ...options }
+  return new DefinitionQueueUpgrader(mergedOptions)
+}
+
+module.exports = serviceFactory

providers/upgrade/defVersionCheck.js

@@ -0,0 +1,47 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const logger = require('../logging/logger')
+const { gte } = require('semver')
+const { get } = require('lodash')
+const EntityCoordinates = require('../../lib/entityCoordinates')
+
+class DefinitionVersionChecker {
+  constructor(options = {}) {
+    this.options = options
+    this.logger = this.options.logger || logger()
+  }
+
+  set currentSchema(schemaVersion) {
+    this._currentSchema = schemaVersion
+  }
+
+  get currentSchema() {
+    return this._currentSchema
+  }
+
+  async validate(definition) {
+    if (!this._currentSchema) throw new Error('Current schema version is not set')
+    const defSchemaVersion = get(definition, '_meta.schemaVersion')
+    this.logger.debug(`Definition version: %s, Current schema version: %s `, defSchemaVersion, this._currentSchema, {
+      coordinates: DefinitionVersionChecker.getCoordinates(definition)
+    })
+    if (defSchemaVersion && gte(defSchemaVersion, this._currentSchema)) return definition
+  }
+
+  async initialize() {
+    //do nothing for initialization
+  }
+
+  setupProcessing() {
+    //do nothing for set up processing
+  }
+
+  static getCoordinates(definition) {
+    return definition?.coordinates && EntityCoordinates.fromObject(definition.coordinates).toString()
+  }
+}
+
+const factory = options => new DefinitionVersionChecker(options)
+
+module.exports = { DefinitionVersionChecker, factory }

providers/upgrade/memoryQueueConfig.js

@@ -0,0 +1,14 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const MemoryQueue = require('../queueing/memoryQueue')
+
+const encodedMessageQueueFactory = opts => {
+  const defaultOpts = {
+    decoder: text => Buffer.from(text, 'base64').toString('utf8')
+  }
+  const mergedOpts = { ...defaultOpts, ...opts }
+  return MemoryQueue(mergedOpts)
+}
+
+module.exports = encodedMessageQueueFactory

providers/upgrade/process.js

@@ -0,0 +1,87 @@
+// (c) Copyright 2024, SAP SE and ClearlyDefined contributors. Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+const { get } = require('lodash')
+const EntityCoordinates = require('../../lib/entityCoordinates')
+const { factory } = require('./defVersionCheck')
+const Cache = require('../caching/memory')
+
+class QueueHandler {
+  constructor(queue, logger, messageHandler = { processMessage: async () => {} }) {
+    this._queue = queue
+    this.logger = logger
+    this._messageHandler = messageHandler
+  }
+
+  async work(once) {
+    let isQueueEmpty = true
+    try {
+      const messages = await this._queue.dequeueMultiple()
+      if (messages && messages.length > 0) isQueueEmpty = false
+      const results = await Promise.allSettled(
+        messages.map(async message => {
+          await this._messageHandler.processMessage(message)
+          await this._queue.delete(message)
+        })
+      )
+      results.filter(result => result.status === 'rejected').forEach(result => this.logger.error(result.reason))
+    } catch (error) {
+      this.logger.error(error)
+    } finally {
+      if (!once) setTimeout(this.work.bind(this), isQueueEmpty ? 10000 : 0)
+    }
+  }
+}
+
+class DefinitionUpgrader {
+  static defaultTtlSeconds = 60 * 5 /* 5 mins */
+  static delayInMSeconds = 500
+
+  constructor(
+    definitionService,
+    logger,
+    defVersionChecker,
+    cache = Cache({ defaultTtlSeconds: DefinitionUpgrader.defaultTtlSeconds })
+  ) {
+    this.logger = logger
+    this._definitionService = definitionService
+    this._defVersionChecker = defVersionChecker
+    this._defVersionChecker.currentSchema = definitionService.currentSchema
+    this._upgradeLock = cache
+  }
+
+  async processMessage(message) {
+    let coordinates = get(message, 'data.coordinates')
+    if (!coordinates) return
+    coordinates = EntityCoordinates.fromObject(coordinates)
+
+    while (this._upgradeLock.get(coordinates.toString())) {
+      await new Promise(resolve => setTimeout(resolve, DefinitionUpgrader.delayInMSeconds))
+    }
+    try {
+      this._upgradeLock.set(coordinates.toString(), true)
+      await this._upgradeIfNecessary(coordinates)
+    } finally {
+      this._upgradeLock.delete(coordinates.toString())
+    }
+  }
+
+  async _upgradeIfNecessary(coordinates) {
+    const existing = await this._definitionService.getStored(coordinates)
+    let result = await this._defVersionChecker.validate(existing)
+    if (!result) {
+      await this._definitionService.computeStoreAndCurate(coordinates)
+      this.logger.info(`Handled definition update for ${coordinates.toString()}`)
+    } else {
+      this.logger.debug(`Skipped definition update for ${coordinates.toString()}`)
+    }
+  }
+}
+
+function setup(_queue, _definitionService, _logger, once = false, _defVersionChecker = factory({ logger: _logger })) {
+  const defUpgrader = new DefinitionUpgrader(_definitionService, _logger, _defVersionChecker)
+  const queueHandler = new QueueHandler(_queue, _logger, defUpgrader)
+  return queueHandler.work(once)
+}
+
+module.exports = { DefinitionUpgrader, QueueHandler, setup }