add simple message queue

Author: piatoss3612

packages/background/package.json

@@ -56,6 +56,7 @@
     "ledger-bitcoin": "^0.2.3",
     "long": "^4.0.0",
     "miscreant": "0.3.2",
+    "p-queue": "^6.6.2",
     "pbkdf2": "^3.1.2",
     "utility-types": "^3.10.0"
   },

packages/background/src/index.ts

@@ -293,12 +293,16 @@ export function init(
     keyRingBitcoinService
   );
 
+  const txExecutableMQ =
+    BackgroundTxExecutor.createMessageQueue<BackgroundTxExecutor.TxExecutableEvent>();
+
   const recentSendHistoryService =
     new RecentSendHistory.RecentSendHistoryService(
       storeCreator("recent-send-history"),
       chainsService,
       backgroundTxService,
-      notification
+      notification,
+      txExecutableMQ.publisher
     );
 
   const settingsService = new Settings.SettingsService(
@@ -323,7 +327,8 @@ export function init(
       backgroundTxService,
       backgroundTxEthereumService,
       analyticsService,
-      recentSendHistoryService
+      recentSendHistoryService,
+      txExecutableMQ.subscriber
     );
 
   Interaction.init(router, interactionService);

packages/background/src/recent-send-history/service.ts

@@ -38,6 +38,7 @@ import {
 import { CoinPretty } from "@keplr-wallet/unit";
 import { simpleFetch } from "@keplr-wallet/simple-fetch";
 import { id } from "@ethersproject/hash";
+import { Publisher, TxExecutableEvent } from "../tx-executor/internal";
 
 const SWAP_API_ENDPOINT = process.env["KEPLR_API_ENDPOINT"] ?? "";
 
@@ -70,7 +71,8 @@ export class RecentSendHistoryService {
     protected readonly kvStore: KVStore,
     protected readonly chainsService: ChainsService,
     protected readonly txService: BackgroundTxService,
-    protected readonly notification: Notification
+    protected readonly notification: Notification,
+    protected readonly publisher: Publisher<TxExecutableEvent> // TODO: publish tx executable event when 트래킹 인덱스가 증가되었을 때
   ) {
     makeObservable(this);
   }

packages/background/src/tx-executor/internal.ts

@@ -1,2 +1,3 @@
 export * from "./service";
 export * from "./init";
+export * from "./message-queue";

packages/background/src/tx-executor/message-queue.ts

@@ -0,0 +1,70 @@
+import PQueue from "p-queue";
+
+class MessageQueueCore<T = unknown> {
+  public subscriber: ((msg: T) => Promise<void> | void) | null = null;
+  public buffer: T[] = [];
+  public queue: PQueue;
+  private isFlushing = false;
+
+  constructor(concurrency = 1) {
+    this.queue = new PQueue({ concurrency });
+  }
+
+  enqueue(message: T) {
+    this.buffer.push(message);
+    this.flush();
+  }
+
+  /**
+   * subscriber 설정
+   */
+  setSubscriber(handler: (msg: T) => Promise<void> | void) {
+    this.subscriber = handler;
+    this.flush();
+  }
+
+  private flush() {
+    if (this.isFlushing) return;
+    if (!this.subscriber) return;
+    if (this.buffer.length === 0) return;
+
+    this.isFlushing = true;
+
+    try {
+      console.log("[MessageQueueCore] flush start", this.buffer.length);
+
+      while (this.buffer.length > 0) {
+        const msg = this.buffer.shift()!;
+        this.queue.add(async () => {
+          try {
+            await this.subscriber!(msg);
+          } catch (e) {
+            console.error("[MessageQueueCore] handler error:", e);
+          }
+        });
+      }
+    } finally {
+      this.isFlushing = false;
+    }
+  }
+}
+export class Publisher<T = unknown> {
+  constructor(private core: MessageQueueCore<T>) {}
+
+  publish(message: T) {
+    this.core.enqueue(message);
+  }
+}
+
+export class Subscriber<T = unknown> {
+  constructor(private core: MessageQueueCore<T>) {}
+
+  subscribe(handler: (msg: T) => Promise<void> | void) {
+    this.core.setSubscriber(handler);
+  }
+}
+
+export interface TxExecutableEvent {
+  readonly executionId: string;
+  readonly executableChainIds: string[];
+}

packages/background/src/tx-executor/service.ts

@@ -46,6 +46,7 @@ import {
   calculateCosmosStdFee,
 } from "./utils/cosmos";
 import { fillUnsignedEVMTx } from "./utils/evm";
+import { Subscriber, TxExecutableEvent } from "./internal";
 export class BackgroundTxExecutorService {
   @observable
   protected recentTxExecutionSeq: number = 0;
@@ -61,7 +62,8 @@ export class BackgroundTxExecutorService {
     protected readonly backgroundTxService: BackgroundTxService,
     protected readonly backgroundTxEthereumService: BackgroundTxEthereumService,
     protected readonly analyticsService: AnalyticsService,
-    protected readonly recentSendHistoryService: RecentSendHistoryService
+    protected readonly recentSendHistoryService: RecentSendHistoryService,
+    protected readonly subscriber: Subscriber<TxExecutableEvent>
   ) {
     makeObservable(this);
   }
@@ -103,16 +105,31 @@ export class BackgroundTxExecutorService {
       );
     });
 
-    // TODO: 간단한 메시지 큐를 구현해서 recent send history service에서 multi tx를 처리할 조건이 만족되었을 때
-    // 이 서비스로 메시지를 보내 트랜잭션을 자동으로 실행할 수 있도록 한다. 굳
+    this.subscriber.subscribe(async ({ executionId, executableChainIds }) => {
+      const execution = this.getTxExecution(executionId);
+      if (!execution) {
+        return;
+      }
+
+      const newExecutableChainIds = executableChainIds.filter((chainId) =>
+        execution.executableChainIds.includes(chainId)
+      );
 
-    // CHECK: 현재 활성화되어 있는 vault에서만 실행할 수 있으면 좋을 듯, how? vaultId 변경 감지? how?
-    // CHECK: 굳이 이걸 백그라운드에서 자동으로 실행할 필요가 있을까?
-    // 불러왔는데 pending 상태거나 오래된 실행이면 사실상 이 작업을 이어가는 것이 의미가 있는지 의문이 든다.
-    // for (const execution of this.getRecentDirectTxsExecutions()) {
+      if (newExecutableChainIds.length === 0) {
+        return;
+      }
+
+      // update the executable chain ids
+      for (const chainId of newExecutableChainIds) {
+        execution.executableChainIds.push(chainId);
+      }
 
-    //   this.executeDirectTxs(execution.id);
-    // }
+      // cause new executable chain ids are available, resume the execution
+
+      // CHECK: 현재 활성화되어 있는 vault에서만 실행할 수 있으면 좋을 듯, how? vaultId 변경 감지? how?
+      // 불러왔는데 pending 상태거나 오래된 실행이면 사실상 이 작업을 이어가는 것이 의미가 있는지 의문이 든다.
+      await this.executeTxs(executionId);
+    });
   }
 
   /**
@@ -225,10 +242,17 @@ export class BackgroundTxExecutorService {
       throw new KeplrError("direct-tx-executor", 105, "Execution not found");
     }
 
+    if (execution.status === TxExecutionStatus.PROCESSING) {
+      throw new KeplrError(
+        "direct-tx-executor",
+        108,
+        "Execution is already processing"
+      );
+    }
+
     // Only pending/processing/blocked executions can be executed
     const needResume =
       execution.status === TxExecutionStatus.PENDING ||
-      execution.status === TxExecutionStatus.PROCESSING ||
       execution.status === TxExecutionStatus.BLOCKED;
     if (!needResume) {
       return execution.status;

yarn.lock

@@ -5857,6 +5857,7 @@ __metadata:
     ledger-bitcoin: ^0.2.3
     long: ^4.0.0
     miscreant: 0.3.2
+    p-queue: ^6.6.2
     pbkdf2: ^3.1.2
     utility-types: ^3.10.0
   peerDependencies: