re-indexing old DDOs

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ocean-node",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "Ocean Node is used to run all core services in the Ocean stack",
   "author": "Ocean Protocol Foundation",
   "license": "Apache-2.0",
@@ -107,7 +107,6 @@
     "rdf-validate-shacl": "^0.5.5",
     "rdflib": "^2.2.33",
     "shacl-engine": "^0.1.2",
-    "sinon": "^17.0.1",
     "sqlite3": "^5.1.7",
     "stream-concat": "^1.0.0",
     "ts-node": "^10.9.1",
@@ -127,6 +126,7 @@
     "@types/mocha": "^10.0.4",
     "@types/node": "^20.14.2",
     "@types/node-cron": "^3.0.11",
+    "@types/sinon": "^17.0.4",
     "@typescript-eslint/eslint-plugin": "^6.8.0",
     "@typescript-eslint/parser": "^6.8.0",
     "auto-changelog": "^2.4.0",
@@ -141,6 +141,7 @@
     "nyc": "^17.1.0",
     "prettier": "^3.0.3",
     "release-it": "^17.6.0",
+    "sinon": "^19.0.2",
     "tsx": "^3.12.8"
   },
   "release-it": {

src/components/Indexer/index.ts

@@ -17,6 +17,7 @@ import { buildJobIdentifier } from './utils.js'
 import { create256Hash } from '../../utils/crypt.js'
 import { isReachableConnection } from '../../utils/database.js'
 import { sleep } from '../../utils/util.js'
+import { isReindexingNeeded } from './version.js'
 
 // emmit events for node
 export const INDEXER_DDO_EVENT_EMITTER = new EventEmitter()
@@ -35,13 +36,19 @@ export class OceanIndexer {
   private networks: RPCS
   private supportedChains: string[]
   private workers: Record<string, Worker> = {}
+  private MIN_REQUIRED_VERSION = '0.2.2'
 
   constructor(db: Database, supportedNetworks: RPCS) {
     this.db = db
     this.networks = supportedNetworks
     this.supportedChains = Object.keys(supportedNetworks)
     INDEXING_QUEUE = []
     this.startThreads()
+
+    // Check if reindexing is needed
+    this.checkAndTriggerReindexing().catch((error) => {
+      INDEXER_LOGGER.error(`Error during version check and reindexing: ${error.message}`)
+    })
   }
 
   public getSupportedNetworks(): RPCS {
@@ -405,4 +412,50 @@ export class OceanIndexer {
       }
     }
   }
+
+  /**
+   * Checks if reindexing is needed and triggers it for all chains
+   */
+  public async checkAndTriggerReindexing(): Promise<void> {
+    const currentVersion = process.env.npm_package_version
+    const dbActive = this.getDatabase()
+    if (!dbActive || !(await isReachableConnection(dbActive.getConfig().url))) {
+      INDEXER_LOGGER.error(`Giving up reindexing. DB is not online!`)
+      return
+    }
+    const dbVersion = await dbActive.indexer.getNodeVersion()
+
+    INDEXER_LOGGER.info(
+      `Node version check: Current=${currentVersion}, DB=${
+        dbVersion || 'not set'
+      }, Min Required=${this.MIN_REQUIRED_VERSION}`
+    )
+
+    if (isReindexingNeeded(currentVersion, dbVersion, this.MIN_REQUIRED_VERSION)) {
+      INDEXER_LOGGER.info(
+        `Reindexing needed: DB version ${
+          dbVersion || 'not set'
+        } is older than minimum required ${this.MIN_REQUIRED_VERSION}`
+      )
+
+      // Reindex all chains
+      for (const chainID of this.supportedChains) {
+        const chainIdNum = Number(chainID)
+        INDEXER_LOGGER.info(`Triggering reindexing for chain ${chainIdNum}`)
+        const job = await this.resetCrawling(chainIdNum)
+        if (!job || job.status === CommandStatus.FAILURE) {
+          INDEXER_LOGGER.error(
+            `Reindex chain job for ${chainIdNum} failed. Please retry reindexChanin command manually for this chain.`
+          )
+          continue
+        }
+      }
+
+      // Update the version in the database
+      await dbActive.indexer.setNodeVersion(currentVersion)
+      INDEXER_LOGGER.info(`Updated node version in database to ${currentVersion}`)
+    } else {
+      INDEXER_LOGGER.info('No reindexing needed based on version check')
+    }
+  }
 }