Merge pull request #1132 from BoltzExchange/interval-amount-batch-claim

Interval and amount batch claim

Author: michael1011

lib/Config.ts

@@ -182,6 +182,11 @@ type SwapConfig = {
   cltvDelta: number;
   sweepAmountTrigger?: number;
 
+  scheduleAmountTrigger?: {
+    interval: string;
+    threshold: number;
+  };
+
   minSwapSizeMultipliers?: MinSwapSizeMultipliersConfig;
 
   overpayment?: OverPaymentConfig;

lib/service/cooperative/DeferredClaimer.ts

@@ -1,6 +1,4 @@
 import AsyncLock from 'async-lock';
-import type { Job } from 'node-schedule';
-import { scheduleJob } from 'node-schedule';
 import type { SwapConfig } from '../../Config';
 import type { ClaimDetails, LiquidClaimDetails } from '../../Core';
 import { calculateTransactionFee, constructClaimTransaction } from '../../Core';
@@ -50,6 +48,8 @@ import CoopSignerBase, {
 } from './CoopSignerBase';
 import AmountTrigger from './triggers/AmountTrigger';
 import ExpiryTrigger from './triggers/ExpiryTrigger';
+import IntervalTrigger from './triggers/IntervalTrigger';
+import ScheduledAmountTrigger from './triggers/ScheduledAmountTrigger';
 import type SweepTrigger from './triggers/SweepTrigger';
 
 type AnySwapWithPreimage<T extends AnySwap> = SwapToClaim<T> & {
@@ -88,7 +88,6 @@ class DeferredClaimer extends CoopSignerBase<{
   >();
   private readonly sweepTriggers: SweepTrigger[];
 
-  private batchClaimSchedule?: Job;
   private disableCooperative = false;
 
   constructor(
@@ -116,6 +115,21 @@ class DeferredClaimer extends CoopSignerBase<{
         this.pendingSweepsValues,
         this.config.sweepAmountTrigger,
       ),
+      new IntervalTrigger(
+        this.logger,
+        this.config.batchClaimInterval,
+        async () => {
+          await this.sweep();
+        },
+      ),
+      new ScheduledAmountTrigger(
+        this.logger,
+        this.config.scheduleAmountTrigger,
+        this.pendingSweepsValues,
+        async (symbol: string) => {
+          await this.sweepSymbol(symbol);
+        },
+      ),
     ];
   }
 
@@ -128,26 +142,20 @@ class DeferredClaimer extends CoopSignerBase<{
       `Using deferred claims for: ${this.config.deferredClaimSymbols.join(', ')}`,
     );
     this.logger.verbose(
-      `Batch claim interval: ${this.config.batchClaimInterval} with expiry tolerance of ${this.config.expiryTolerance} minutes`,
+      `Expiry tolerance: ${this.config.expiryTolerance} minutes`,
     );
 
     try {
       await this.batchClaimLeftovers();
     } catch (e) {
       this.logger.error(`Could not sweep leftovers: ${formatError(e)}`);
     }
-
-    this.batchClaimSchedule = scheduleJob(
-      this.config.batchClaimInterval,
-      async () => {
-        await this.sweep();
-      },
-    );
   };
 
   public close = () => {
-    this.batchClaimSchedule?.cancel();
-    this.batchClaimSchedule = undefined;
+    for (const trigger of this.sweepTriggers) {
+      trigger.close();
+    }
   };
 
   public pendingSweeps = () => {

lib/service/cooperative/triggers/AmountTrigger.ts

@@ -43,6 +43,8 @@ class AmountTrigger extends SweepTrigger {
     );
     return toSweep >= this.sweepAmountTrigger;
   };
+
+  public close = () => {};
 }
 
 export default AmountTrigger;

lib/service/cooperative/triggers/ExpiryTrigger.ts

@@ -30,6 +30,8 @@ class ExpiryTrigger extends SweepTrigger {
     return minutesLeft <= this.expiryTolerance;
   };
 
+  public close = () => {};
+
   private getBlockHeight = async (chainCurrency: string) => {
     const currency = this.currencies.get(chainCurrency);
     if (!currency) {

lib/service/cooperative/triggers/IntervalTrigger.ts

@@ -0,0 +1,43 @@
+import type { Job } from 'node-schedule';
+import { scheduleJob } from 'node-schedule';
+import type Logger from '../../../Logger';
+import { formatError } from '../../../Utils';
+import SweepTrigger from './SweepTrigger';
+
+class IntervalTrigger extends SweepTrigger {
+  private batchClaimSchedule?: Job;
+
+  constructor(
+    private readonly logger: Logger,
+    private readonly interval: string,
+    callback: () => Promise<void>,
+  ) {
+    super();
+
+    this.logger.verbose(`Batch claim interval: ${this.interval}`);
+
+    this.batchClaimSchedule = scheduleJob(this.interval, async () => {
+      this.logger.verbose('Batch claim interval triggered');
+      try {
+        await callback();
+      } catch (error) {
+        this.logger.error(
+          `Error in batch claim interval callback: ${formatError(error)}`,
+        );
+      }
+    });
+  }
+
+  public check = async (): Promise<boolean> => {
+    // Interval trigger doesn't check per-swap conditions
+    // It triggers on a schedule instead
+    return false;
+  };
+
+  public close = () => {
+    this.batchClaimSchedule?.cancel();
+    this.batchClaimSchedule = undefined;
+  };
+}
+
+export default IntervalTrigger;

lib/service/cooperative/triggers/ScheduledAmountTrigger.ts

@@ -0,0 +1,89 @@
+import type { Job } from 'node-schedule';
+import { scheduleJob } from 'node-schedule';
+import type { SwapConfig } from '../../../Config';
+import type Logger from '../../../Logger';
+import { formatError } from '../../../Utils';
+import type DeferredClaimer from '../DeferredClaimer';
+import SweepTrigger from './SweepTrigger';
+
+class ScheduledAmountTrigger extends SweepTrigger {
+  private readonly threshold?: number;
+
+  private batchClaimSchedule?: Job;
+
+  constructor(
+    private readonly logger: Logger,
+    config: SwapConfig['scheduleAmountTrigger'] | undefined,
+    private readonly pendingValues: typeof DeferredClaimer.prototype.pendingSweepsValues,
+    private readonly onTrigger: (symbol: string) => Promise<void>,
+  ) {
+    super();
+
+    if (config === undefined) {
+      this.logger.warn('Scheduled amount trigger not set');
+      return;
+    }
+
+    if (config.threshold === undefined) {
+      throw new Error('scheduleAmountTrigger.threshold is required');
+    }
+
+    if (config.interval === undefined) {
+      throw new Error('scheduleAmountTrigger.interval is required');
+    }
+
+    this.threshold = config.threshold;
+    this.logger.verbose(
+      `Scheduled amount trigger: >= ${this.threshold} every ${config.interval}`,
+    );
+
+    this.batchClaimSchedule = scheduleJob(config.interval, async () => {
+      try {
+        await this.checkAndTrigger();
+      } catch (error) {
+        this.logger.error(
+          `Error in scheduled amount trigger check: ${formatError(error)}`,
+        );
+      }
+    });
+  }
+
+  public check = async (): Promise<boolean> => {
+    // Scheduled amount trigger doesn't check per-swap conditions
+    // It triggers on a schedule instead
+    return false;
+  };
+
+  public close = () => {
+    this.batchClaimSchedule?.cancel();
+    this.batchClaimSchedule = undefined;
+  };
+
+  private checkAndTrigger = async () => {
+    if (this.threshold === undefined) {
+      return;
+    }
+
+    for (const [symbol, pendingValues] of this.pendingValues()) {
+      const totalAmount = pendingValues.reduce(
+        (acc, value) => acc + value.onchainAmount,
+        0,
+      );
+
+      if (totalAmount >= this.threshold) {
+        this.logger.verbose(
+          `Scheduled amount trigger for ${symbol}: ${totalAmount} >= ${this.threshold}`,
+        );
+        try {
+          await this.onTrigger(symbol);
+        } catch (triggerError) {
+          this.logger.error(
+            `Error triggering scheduled amount sweep for ${symbol}: ${formatError(triggerError)}`,
+          );
+        }
+      }
+    }
+  };
+}
+
+export default ScheduledAmountTrigger;

lib/service/cooperative/triggers/SweepTrigger.ts

@@ -6,6 +6,8 @@ abstract class SweepTrigger {
     chainCurrency: string,
     swap: Swap | ChainSwapInfo,
   ): Promise<boolean>;
+
+  public abstract close(): void;
 }
 
 export default SweepTrigger;

test/integration/service/cooperative/DeferredClaimer.spec.ts

@@ -455,7 +455,6 @@ describe('DeferredClaimer', () => {
       expect(ChainSwapRepository.getChainSwapsClaimable).toHaveBeenCalledTimes(
         1,
       );
-      expect(claimer['batchClaimSchedule']).not.toBeUndefined();
     });
 
     test('should not crash when batch claim of leftovers fails', async () => {
@@ -470,12 +469,6 @@ describe('DeferredClaimer', () => {
     });
   });
 
-  test('should close', () => {
-    expect(claimer['batchClaimSchedule']).not.toBeUndefined();
-    claimer.close();
-    expect(claimer['batchClaimSchedule']).toBeUndefined();
-  });
-
   describe('deferClaim', () => {
     test('should defer claim transactions of Submarine Swaps', async () => {
       const swap = {

test/integration/swap/UtxoNursery.spec.ts

@@ -186,6 +186,8 @@ describe('UtxoNursery', () => {
       ] as ClnPendingPaymentTracker
     ).stop();
 
+    swapManager.deferredClaimer.close();
+
     sidecar.disconnect();
     await Sidecar.stop();
 

test/unit/service/cooperative/triggers/IntervalTrigger.spec.ts

@@ -0,0 +1,95 @@
+import { scheduleJob } from 'node-schedule';
+import type Logger from '../../../../../lib/Logger';
+import IntervalTrigger from '../../../../../lib/service/cooperative/triggers/IntervalTrigger';
+
+jest.mock('node-schedule');
+
+describe('IntervalTrigger', () => {
+  const mockLogger = {
+    verbose: jest.fn(),
+  } as unknown as Logger;
+
+  const mockCallback = jest.fn().mockResolvedValue(undefined);
+  const interval = '*/5 * * * *';
+
+  let mockJob: any;
+  let trigger: IntervalTrigger;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+
+    mockJob = {
+      cancel: jest.fn(),
+    };
+
+    (scheduleJob as jest.Mock).mockReturnValue(mockJob);
+  });
+
+  describe('constructor', () => {
+    test('should log the batch claim interval', () => {
+      trigger = new IntervalTrigger(mockLogger, interval, mockCallback);
+
+      expect(mockLogger.verbose).toHaveBeenCalledTimes(1);
+      expect(mockLogger.verbose).toHaveBeenCalledWith(
+        `Batch claim interval: ${interval}`,
+      );
+    });
+
+    test('should schedule a job with the provided interval and callback', () => {
+      trigger = new IntervalTrigger(mockLogger, interval, mockCallback);
+
+      expect(scheduleJob).toHaveBeenCalledTimes(1);
+      expect(scheduleJob).toHaveBeenCalledWith(interval, expect.any(Function));
+    });
+
+    test('should execute the callback when the scheduled job runs', async () => {
+      trigger = new IntervalTrigger(mockLogger, interval, mockCallback);
+
+      const scheduledFunction = (scheduleJob as jest.Mock).mock.calls[0][1];
+
+      await scheduledFunction();
+
+      expect(mockCallback).toHaveBeenCalledTimes(1);
+    });
+
+    test('should work with different interval formats', () => {
+      const cronInterval = '0 */2 * * *';
+      trigger = new IntervalTrigger(mockLogger, cronInterval, mockCallback);
+
+      expect(scheduleJob).toHaveBeenCalledWith(
+        cronInterval,
+        expect.any(Function),
+      );
+    });
+  });
+
+  describe('check', () => {
+    beforeEach(() => {
+      trigger = new IntervalTrigger(mockLogger, interval, mockCallback);
+    });
+
+    test('should always return false', async () => {
+      const result = await trigger.check();
+      expect(result).toBe(false);
+    });
+  });
+
+  describe('close', () => {
+    beforeEach(() => {
+      trigger = new IntervalTrigger(mockLogger, interval, mockCallback);
+    });
+
+    test('should cancel the scheduled job', () => {
+      trigger.close();
+
+      expect(mockJob.cancel).toHaveBeenCalledTimes(1);
+    });
+
+    test('should handle close being called multiple times', () => {
+      trigger.close();
+      trigger.close();
+
+      expect(mockJob.cancel).toHaveBeenCalledTimes(1);
+    });
+  });
+});