sqlite writing queue

packages/data-store-typeorm/package.json

@@ -18,6 +18,7 @@
   },
   "dependencies": {
     "@docknetwork/wallet-sdk-wasm": "^0.4.19",
+    "async-mutex": "^0.5.0",
     "uuid": "^8.3.2"
   },
   "devDependencies": {

packages/data-store-typeorm/src/entities/document/create-document.ts

@@ -1,10 +1,17 @@
-import {ContextProps, DataStoreEvents, WalletDocument} from '@docknetwork/wallet-sdk-data-store/src/types';
+import {
+  ContextProps,
+  DataStoreEvents,
+  WalletDocument,
+} from '@docknetwork/wallet-sdk-data-store/src/types';
 import {v4 as uuid} from 'uuid';
 import {DocumentEntity} from './document.entity';
 import {getOrCreateDocumentTypes, saveOptions} from './helpers';
 import {getDocumentById} from './get-document-by-id';
 import {logger} from '../../logger';
-import { getDataSource } from '../../helpers';
+import {getDataSource} from '../../helpers';
+import {Mutex} from 'async-mutex';
+
+export const writeMutex = new Mutex();
 
 /**
  * Create document
@@ -19,64 +26,66 @@ export async function createDocument({
   json: any;
   options?: any;
 }): Promise<WalletDocument> {
-  logger.debug(`Creating document with id ${json.id}...`);
-  if (json.id) {
-    const existingDocument = await getDocumentById({
-      dataStore,
-      id: json.id,
-    });
+  return writeMutex.runExclusive(async () => {
+    logger.debug(`Creating document with id ${json.id}...`);
+    if (json.id) {
+      const existingDocument = await getDocumentById({
+        dataStore,
+        id: json.id,
+      });
 
-    if (existingDocument) {
-      logger.debug(`Document with id ${json.id} already exists`);
-      throw new Error(`Document with id ${json.id} already exists`);
+      if (existingDocument) {
+        logger.debug(`Document with id ${json.id} already exists`);
+        throw new Error(`Document with id ${json.id} already exists`);
+      }
     }
-  }
 
-  const _typeRel = await getOrCreateDocumentTypes({
-    dataStore,
-    types: json.type,
-  });
+    const _typeRel = await getOrCreateDocumentTypes({
+      dataStore,
+      types: json.type,
+    });
 
-  if (!json.id) {
-    json.id = uuid();
-  }
+    if (!json.id) {
+      json.id = uuid();
+    }
 
-  let networkId;
+    let networkId;
 
-  if (json._networkId) {
-    networkId = json._networkId;
-    delete json._networkId;
-  } else {
-    const resolution = await dataStore.resolveDocumentNetwork({
-      document: json,
-      dataStore,
-    });
+    if (json._networkId) {
+      networkId = json._networkId;
+      delete json._networkId;
+    } else {
+      const resolution = await dataStore.resolveDocumentNetwork({
+        document: json,
+        dataStore,
+      });
 
-    networkId = resolution.networkId;
-  }
+      networkId = resolution.networkId;
+    }
 
-  const entity: DocumentEntity = {
-    networkId,
-    id: json.id,
-    type: json.type,
-    _typeRel,
-    correlation: json.correlation || [],
-    data: JSON.stringify(json),
-  };
+    const entity: DocumentEntity = {
+      networkId,
+      id: json.id,
+      type: json.type,
+      _typeRel,
+      correlation: json.correlation || [],
+      data: JSON.stringify(json),
+    };
 
-  const db = getDataSource(dataStore);
+    const db = getDataSource(dataStore);
 
-  const repository = db.getRepository(DocumentEntity);
+    const repository = db.getRepository(DocumentEntity);
 
-  const result = await repository.save(entity, saveOptions);
+    const result = await repository.save(entity, saveOptions);
 
-  if (!options?.stopPropagation) {
-    dataStore.events.emit(DataStoreEvents.DocumentCreated, json);
-  } else {
-    console.log('stopPropagation is true')
-  }
+    if (!options?.stopPropagation) {
+      dataStore.events.emit(DataStoreEvents.DocumentCreated, json);
+    } else {
+      console.log('stopPropagation is true');
+    }
 
-  logger.debug(`Document added to the wallet`);
+    logger.debug(`Document added to the wallet`);
 
-  return result;
+    return result;
+  });
 }

packages/data-store-typeorm/src/entities/document/update-document.ts

@@ -1,8 +1,12 @@
-import { ContextProps, WalletDocument } from '@docknetwork/wallet-sdk-data-store/src/types';
-import { saveOptions, toDocumentEntity } from './helpers';
-import { DocumentEntity } from './document.entity';
-import { logger } from '../../logger';
-import { getDataSource } from '../../helpers';
+import {
+  ContextProps,
+  WalletDocument,
+} from '@docknetwork/wallet-sdk-data-store/src/types';
+import {saveOptions, toDocumentEntity} from './helpers';
+import {DocumentEntity} from './document.entity';
+import {logger} from '../../logger';
+import {getDataSource} from '../../helpers';
+import {writeMutex} from './create-document';
 
 /**
  * Update document
@@ -12,18 +16,23 @@ import { getDataSource } from '../../helpers';
 export async function updateDocument({
   dataStore,
   document,
-}: ContextProps & {document: WalletDocument, options?: any}): Promise<WalletDocument> {
-  logger.debug(`Updating document with id ${document.id}`);
+}: ContextProps & {
+  document: WalletDocument;
+  options?: any;
+}): Promise<WalletDocument> {
+  return writeMutex.runExclusive(async () => {
+    logger.debug(`Updating document with id ${document.id}`);
 
-  const db = getDataSource(dataStore);
-  const repository = db.getRepository(DocumentEntity);
-  const entity = await toDocumentEntity({
-    dataStore,
-    document,
-  });
-  await repository.save(entity, saveOptions);
+    const db = getDataSource(dataStore);
+    const repository = db.getRepository(DocumentEntity);
+    const entity = await toDocumentEntity({
+      dataStore,
+      document,
+    });
+    await repository.save(entity, saveOptions);
 
-  dataStore.events.emit('DocumentUpdated', document);
+    dataStore.events.emit('DocumentUpdated', document);
 
-  return document;
+    return document;
+  });
 }

yarn.lock

@@ -8500,6 +8500,13 @@ astral-regex@^2.0.0:
   resolved "https://registry.yarnpkg.com/astral-regex/-/astral-regex-2.0.0.tgz#483143c567aeed4785759c0865786dc77d7d2e31"
   integrity sha512-Z7tMw1ytTXt5jqMcOP+OQteU1VuNK9Y02uuJtKQ1Sv69jXQKKg5cibLwGJow8yzZP+eAc18EmLGPal0bp36rvQ==
 
+async-mutex@^0.5.0:
+  version "0.5.0"
+  resolved "https://registry.yarnpkg.com/async-mutex/-/async-mutex-0.5.0.tgz#353c69a0b9e75250971a64ac203b0ebfddd75482"
+  integrity sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==
+  dependencies:
+    tslib "^2.4.0"
+
 async@^3.2.3:
   version "3.2.4"
   resolved "https://registry.yarnpkg.com/async/-/async-3.2.4.tgz#2d22e00f8cddeb5fde5dd33522b56d1cf569a81c"