fix: reduce redis data size, exclude UTXOs data from job return value

.gitignore

@@ -31,3 +31,4 @@ data
 redis-data
 dump.rdb
 .envrc
+.env.signet

src/plugins/cron.ts

@@ -90,9 +90,9 @@ export default fp(async (fastify) => {
           onCompleted: async (job) => {
             fastify.log.info(`[UTXOSyncer] job completed: ${job.id}`);
             if (env.RGBPP_COLLECT_DATA_CACHE_ENABLE) {
-              const { btcAddress, utxos } = job.returnvalue;
+              const { btcAddress } = job.data;
               const rgbppCollector: RgbppCollector = fastify.container.resolve('rgbppCollector');
-              await rgbppCollector.enqueueCollectJob(btcAddress, utxos, true);
+              await rgbppCollector.enqueueCollectJob(btcAddress, true);
             }
           },
         });

src/routes/bitcoin/address.ts

@@ -53,7 +53,7 @@ const addressRoutes: FastifyPluginCallback<Record<never, never>, Server, ZodType
         no_cache === 'true',
       );
       if (env.RGBPP_COLLECT_DATA_CACHE_ENABLE) {
-        await fastify.rgbppCollector.enqueueCollectJob(address, utxos);
+        await fastify.rgbppCollector.enqueueCollectJob(address);
       }
 
       const rgbppUtxoMap = rgbppUtxoCellsPairs.reduce((map, { utxo }) => {
@@ -140,7 +140,7 @@ const addressRoutes: FastifyPluginCallback<Record<never, never>, Server, ZodType
           ? await fastify.rgbppCollector.getRgbppUtxoCellsPairs(address, utxos, no_cache === 'true')
           : [];
       if (env.RGBPP_COLLECT_DATA_CACHE_ENABLE) {
-        await fastify.rgbppCollector.enqueueCollectJob(address, utxos);
+        await fastify.rgbppCollector.enqueueCollectJob(address);
       }
       const rgbppUtxoSet = new Set(rgbppUtxoCellsPairs.map((pair) => pair.utxo.txid + ':' + pair.utxo.vout));
 

src/routes/rgbpp/address.ts

@@ -63,7 +63,7 @@ const addressRoutes: FastifyPluginCallback<Record<never, never>, Server, ZodType
       no_cache === 'true',
     );
     if (env.RGBPP_COLLECT_DATA_CACHE_ENABLE) {
-      await fastify.rgbppCollector.enqueueCollectJob(btc_address, utxos);
+      await fastify.rgbppCollector.enqueueCollectJob(btc_address);
     }
     const cells = rgbppUtxoCellsPairs.map((pair) => pair.cells).flat();
     return cells;

src/services/base/queue-worker.ts

@@ -41,7 +41,7 @@ export default abstract class BaseQueueWorker<T, R> {
     );
   }
 
-  abstract process(job: Job<T>): Promise<R>;
+  abstract process(job: Job<T>): Promise<R | void>;
 
   /**
    * Add a job to the queue

src/services/rgbpp.ts

@@ -25,7 +25,6 @@ export type RgbppUtxoCellsPair = {
 
 interface IRgbppCollectRequest {
   btcAddress: string;
-  utxos: UTXO[];
 }
 
 interface IRgbppCollectJobReturn {
@@ -80,12 +79,9 @@ export default class RgbppCollector extends BaseQueueWorker<IRgbppCollectRequest
    * @param err - the error
    */
   private captureJobExceptionToSentryScope(job: Job<IRgbppCollectRequest>, err: Error) {
-    const { btcAddress, utxos } = job.data;
+    const { btcAddress } = job.data;
     Sentry.withScope((scope) => {
       scope.setTag('btcAddress', btcAddress);
-      scope.setContext('utxos', {
-        utxos: JSON.stringify(utxos),
-      });
       this.cradle.logger.error(err);
       scope.captureException(err);
     });
@@ -235,18 +231,21 @@ export default class RgbppCollector extends BaseQueueWorker<IRgbppCollectRequest
    * Enqueue a collect job to the queue
    * @param utxos - the utxos to collect
    */
-  public async enqueueCollectJob(
-    btcAddress: string,
-    utxos: UTXO[],
-    allowDuplicate?: boolean,
-  ): Promise<Job<IRgbppCollectRequest>> {
+  public async enqueueCollectJob(btcAddress: string, allowDuplicate?: boolean): Promise<Job<IRgbppCollectRequest>> {
     let jobId = btcAddress;
     if (allowDuplicate) {
       // add a timestamp to the job id to allow duplicate jobs
       // used for the case that the utxos are updated
       jobId = `${btcAddress}:${Date.now()}`;
     }
-    return this.addJob(jobId, { btcAddress, utxos });
+    return this.addJob(
+      jobId,
+      { btcAddress },
+      {
+        removeOnComplete: true,
+        removeOnFail: true,
+      },
+    );
   }
 
   /**
@@ -256,10 +255,10 @@ export default class RgbppCollector extends BaseQueueWorker<IRgbppCollectRequest
    */
   public async process(job: Job<IRgbppCollectRequest>) {
     try {
-      const { btcAddress, utxos } = job.data;
+      const { btcAddress } = job.data;
+      const utxos = await this.cradle.utxoSyncer.getUtxosByAddress(btcAddress);
       const pairs = await this.collectRgbppUtxoCellsPairs(utxos);
-      const data = await this.saveRgbppUtxoCellsPairsToCache(btcAddress, pairs);
-      return data;
+      await this.saveRgbppUtxoCellsPairsToCache(btcAddress, pairs);
     } catch (e) {
       const { message, stack } = e as Error;
       const error = new RgbppCollectorError(message);

src/services/utxo.ts

@@ -151,6 +151,8 @@ export default class UTXOSyncer extends BaseQueueWorker<IUTXOSyncRequest, IUTXOS
           // https://github.com/taskforcesh/bullmq/blob/cce0774cffcee591407eee4d4530daa37aab3eca/src/classes/repeat.ts#L51
           endDate: Date.now() + this.cradle.env.UTXO_SYNC_REPEAT_EXPRIED_DURATION,
         },
+        removeOnComplete: true,
+        removeOnFail: true,
       },
     );
   }
@@ -167,7 +169,7 @@ export default class UTXOSyncer extends BaseQueueWorker<IUTXOSyncRequest, IUTXOS
     return this.enqueueSyncJobThrottle(btcAddress);
   }
 
-  public async process(job: Job<IUTXOSyncRequest>): Promise<IUTXOSyncJobReturn> {
+  public async process(job: Job<IUTXOSyncRequest>) {
     try {
       const { btcAddress } = job.data;
       if (!validateBitcoinAddress(btcAddress)) {
@@ -189,7 +191,7 @@ export default class UTXOSyncer extends BaseQueueWorker<IUTXOSyncRequest, IUTXOS
 
       const utxos = await this.cradle.bitcoin.getAddressTxsUtxo({ address: btcAddress });
       const data = { btcAddress, utxos, txsHash };
-      return this.dataCache.set(btcAddress, data);
+      await this.dataCache.set(btcAddress, data);
     } catch (e) {
       const { message, stack } = e as Error;
       const error = new UTXOSyncerError(message);