fix: manage concurent initializations (#3132)

---------

Co-authored-by: Farnabaz <farnabaz@gmail.com>

Author: elevatebart

src/module.ts

@@ -244,10 +244,6 @@ async function processCollectionItems(nuxt: Nuxt, collections: ResolvedCollectio
     }
     const collectionHash = hash(collection)
     collectionDump[collection.name] = []
-    // Collection table definition
-    collectionDump[collection.name]!.push(
-      ...generateCollectionTableDefinition(collection, { drop: true }).split('\n'),
-    )
 
     if (!collection.source) {
       continue
@@ -302,14 +298,28 @@ async function processCollectionItems(nuxt: Nuxt, collections: ResolvedCollectio
       }
       // Sort by file name to ensure consistent order
       list.sort((a, b) => String(a[0]).localeCompare(String(b[0])))
-      collectionDump[collection.name]!.push(...list.flatMap(([, sql]) => sql!))
+      const insertQueriesList = list.flatMap(([, sql]) => sql!)
 
-      collectionChecksum[collection.name] = hash(collectionDump[collection.name])
+      // Collection table definition
+      const insertQueriesListWithDefinition = [...generateCollectionTableDefinition(collection, { drop: true }).split('\n'), ...insertQueriesList]
 
+      collectionChecksum[collection.name] = hash(insertQueriesListWithDefinition)
+
+      // we have to start the series of queries
+      // by telling everyone that we are setting up the collection so no
+      // other request start doing the same work and fail
+      // so we create a new entry in the info table saying that it is not ready yet
       collectionDump[collection.name]!.push(
         generateCollectionTableDefinition(infoCollection, { drop: false }),
         `DELETE FROM ${infoCollection.tableName} WHERE id = 'checksum_${collection.name}';`,
-        ...generateCollectionInsert(infoCollection, { id: `checksum_${collection.name}`, version: collectionChecksum[collection.name] }),
+        ...generateCollectionInsert(infoCollection, { id: `checksum_${collection.name}`, version: collectionChecksum[collection.name], ready: false }),
+      )
+
+      collectionDump[collection.name]!.push(...insertQueriesListWithDefinition)
+
+      // and finally when we are finished, we update the info table to say that the init is done
+      collectionDump[collection.name]!.push(
+        `UPDATE ${infoCollection.tableName} SET ready = true WHERE id = 'checksum_${collection.name}'`,
       )
     }
   }

src/runtime/internal/database.server.ts

@@ -45,7 +45,9 @@ export async function checkAndImportDatabaseIntegrity(event: H3Event, collection
   if (checkDatabaseIntegrity[String(collection)] !== false) {
     checkDatabaseIntegrity[String(collection)] = false
     integrityCheckPromise[String(collection)] = integrityCheckPromise[String(collection)] || _checkAndImportDatabaseIntegrity(event, collection, checksums[String(collection)], config)
-      .then((isValid) => { checkDatabaseIntegrity[String(collection)] = !isValid })
+      .then((isValid) => {
+        checkDatabaseIntegrity[String(collection)] = !isValid
+      })
       .catch((error) => {
         console.error('Database integrity check failed', error)
         checkDatabaseIntegrity[String(collection)] = true
@@ -61,13 +63,44 @@ export async function checkAndImportDatabaseIntegrity(event: H3Event, collection
 async function _checkAndImportDatabaseIntegrity(event: H3Event, collection: string, integrityVersion: string, config: RuntimeConfig['content']) {
   const db = loadDatabaseAdapter(config)
 
-  const before = await db.first<{ version: string }>(`select * from ${tables.info} where id = ?`, [`checksum_${collection}`]).catch(() => ({ version: '' }))
+  const before: { version: string, ready: boolean } | null = await db.first<{ version: string, ready: boolean }>(`select * from ${tables.info} where id = ?`, [`checksum_${collection}`]).catch((): null => null)
 
   if (before?.version) {
-    if (before?.version === integrityVersion) {
+    if (before.ready === true && before.version === integrityVersion) {
+      // table is already initialized and ready, use it
+      return true
+    }
+
+    if (before.ready === false && before.version === integrityVersion) {
+      // if another request has already started the initialization of
+      // this version of this collection, wait for it to finish
+      // then respond that the database is ready
+      // NOTE: only wait if the version is the same so if the previous init
+      // was interrupted or has failed, it will not block the new init
+      let iterationCount = 0
+      await new Promise((resolve, reject) => {
+        const interval = setInterval(async () => {
+          const { ready } = await db.first<{ ready: boolean }>(`select ready from ${tables.info} where id = ?`, [`checksum_${collection}`]).catch(() => ({ ready: true }))
+
+          if (ready) {
+            clearInterval(interval)
+            resolve(0)
+          }
+
+          // after timeout is reached, give up and stop the query
+          // it has to be that initialization has failed
+          if (iterationCount++ > 300) {
+            clearInterval(interval)
+            reject(new Error('Waiting for another database initialization timed out'))
+          }
+        }, 1000)
+      }).catch((e) => {
+        throw e
+      })
       return true
     }
-    // Delete old version
+
+    // Delete old version -- checksum exists but does not match with bundled checksum
     await db.exec(`DELETE FROM ${tables.info} WHERE id = ?`, [`checksum_${collection}`])
   }
 

src/utils/collection.ts

@@ -84,10 +84,12 @@ export function resolveCollections(collections: Record<string, DefinedCollection
     schema: z.object({
       id: z.string(),
       version: z.string(),
+      ready: z.boolean(),
     }),
     extendedSchema: z.object({
       id: z.string(),
       version: z.string(),
+      ready: z.boolean(),
     }),
     fields: {},
   }