feat: split data-access read and write for TheGraph (#875)

packages/request-client.js/src/api/request.ts

@@ -382,7 +382,10 @@ export default class Request {
     }
 
     extensionsData.push(
-      declarativePaymentNetwork.createExtensionsDataForDeclareSentPayment({ amount, note }),
+      declarativePaymentNetwork.createExtensionsDataForDeclareSentPayment({
+        amount,
+        note,
+      }),
     );
 
     const parameters: RequestLogicTypes.IAddExtensionsDataParameters = {

packages/request-node/src/request/getChannelsByTopic.ts

@@ -4,7 +4,7 @@ import { StatusCodes } from 'http-status-codes';
 
 const GET_CHANNELS_TIMEOUT = 600000;
 export default class GetChannelHandler {
-  constructor(private logger: LogTypes.ILogger, private dataAccess: DataAccessTypes.IDataAccess) {
+  constructor(private logger: LogTypes.ILogger, private dataAccess: DataAccessTypes.IDataRead) {
     this.handler = this.handler.bind(this);
   }
 

packages/request-node/src/request/getTransactionsByChannelId.ts

@@ -5,7 +5,7 @@ import { StatusCodes } from 'http-status-codes';
 const GET_TRANSACTIONS_TIMEOUT = 600000;
 
 export default class GetTransactionsByChannelIdHandler {
-  constructor(private logger: LogTypes.ILogger, private dataAccess: DataAccessTypes.IDataAccess) {
+  constructor(private logger: LogTypes.ILogger, private dataAccess: DataAccessTypes.IDataRead) {
     this.handler = this.handler.bind(this);
   }
 

packages/request-node/src/request/persistTransaction.ts

@@ -15,7 +15,7 @@ export default class PersistTransactionHandler {
    */
   constructor(
     private confirmedTransactionStore: ConfirmedTransactionStore,
-    private dataAccess: DataAccessTypes.IDataAccess,
+    private dataAccess: DataAccessTypes.IDataWrite,
     private logger: LogTypes.ILogger,
   ) {
     this.handler = this.handler.bind(this);

packages/request-node/src/thegraph/CombinedDataAccess.ts

@@ -0,0 +1,44 @@
+import { DataAccessTypes } from '@requestnetwork/types';
+
+export abstract class CombinedDataAccess implements DataAccessTypes.IDataAccess {
+  constructor(
+    protected reader: DataAccessTypes.IDataRead,
+    protected writer: DataAccessTypes.IDataWrite,
+  ) {
+    this.getTransactionsByChannelId = this.reader.getTransactionsByChannelId.bind(this.reader);
+    this.getChannelsByTopic = this.reader.getChannelsByTopic.bind(this.reader);
+    this.getChannelsByMultipleTopics = this.reader.getChannelsByMultipleTopics.bind(this.reader);
+    this.persistTransaction = this.writer.persistTransaction.bind(this.writer);
+  }
+
+  async initialize(): Promise<void> {
+    await this.reader.initialize();
+    await this.writer.initialize();
+  }
+
+  async close(): Promise<void> {
+    await this.writer.close();
+    await this.reader.close();
+  }
+
+  getTransactionsByChannelId: (
+    channelId: string,
+    updatedBetween?: DataAccessTypes.ITimestampBoundaries | undefined,
+  ) => Promise<DataAccessTypes.IReturnGetTransactions>;
+
+  getChannelsByTopic: (
+    topic: string,
+    updatedBetween?: DataAccessTypes.ITimestampBoundaries | undefined,
+  ) => Promise<DataAccessTypes.IReturnGetChannelsByTopic>;
+  getChannelsByMultipleTopics: (
+    topics: string[],
+    updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+  ) => Promise<DataAccessTypes.IReturnGetChannelsByTopic>;
+  persistTransaction: (
+    transactionData: DataAccessTypes.ITransaction,
+    channelId: string,
+    topics?: string[] | undefined,
+  ) => Promise<DataAccessTypes.IReturnPersistTransaction>;
+
+  abstract _getStatus(): Promise<any>;
+}

packages/request-node/src/thegraph/PendingStore.ts

@@ -0,0 +1,30 @@
+import { DataAccessTypes, StorageTypes } from '@requestnetwork/types';
+
+type PendingItem = {
+  transaction: DataAccessTypes.ITransaction;
+  storageResult: StorageTypes.IAppendResult;
+};
+/**
+ * A simple in-memory store to share state between DataReader and DataWriter
+ * Useful to retrieve a transaction that was just emitted but is not confirmed yet
+ **/
+export class PendingStore {
+  private pending: Record<string, PendingItem> = {};
+
+  /** Gets a pending tx */
+  public get(channelId: string): PendingItem {
+    return this.pending[channelId];
+  }
+
+  public add(
+    channelId: string,
+    transaction: DataAccessTypes.ITransaction,
+    storageResult: StorageTypes.IAppendResult,
+  ): void {
+    this.pending[channelId] = { transaction, storageResult };
+  }
+
+  public remove(channelId: string): void {
+    delete this.pending[channelId];
+  }
+}