feat: create schema and service for managing UTXOs

packages/worker/src/index.ts

@@ -8,7 +8,12 @@ import AssetCron from "./queues/assetsValue/scheduler";
 import assetQueue from "./queues/assetsValue/queue";
 import { MongoDatabase } from "./clients/mongoClient";
 import { PsqlClient } from "./clients";
-import { userBlockSyncQueue, userLogoutSyncQueue, UserBlockSyncCron } from "./queues/userBlockSync";
+import {
+  userBlockSyncQueue,
+  userLogoutSyncQueue,
+  UserBlockSyncCron,
+} from "./queues/userBlockSync";
+import { GaslessUtxoCleanup } from "@/queues/gaslessUtxos/gaslessUtxoCleanup";
 
 const {
   WORKER_PORT,
@@ -73,6 +78,7 @@ PsqlClient.connect();
 BalanceCron.create();
 AssetCron.create();
 UserBlockSyncCron.create();
+GaslessUtxoCleanup.start();
 
 app.listen(WORKER_PORT ?? 3063, () =>
   console.log(`Server running on ${WORKER_PORT}`)

packages/worker/src/queues/gaslessUtxos/constants.ts

@@ -0,0 +1,4 @@
+export const COLLECTION_GASLESS_UTXOS = "gasless_utxos";
+export const DEFAULT_TTL_SECONDS = 60 * 60;
+export const CLEANUP_INTERVAL_MS = 60 * 1000;
+export const RESERVE_TTLS_MS = 5 * 60 * 1000;

packages/worker/src/queues/gaslessUtxos/gaslessUtxoCleanup.ts

@@ -0,0 +1,43 @@
+import { MongoDatabase } from "@/clients/mongoClient";
+import {
+  COLLECTION_GASLESS_UTXOS,
+  CLEANUP_INTERVAL_MS,
+} from "@/queues/gaslessUtxos/constants";
+import { GaslessUtxo } from "@/queues/gaslessUtxos/types";
+import { gaslessUtxosCollection } from "@/queues/gaslessUtxos";
+
+export class GaslessUtxoCleanup {
+  private static instance?: GaslessUtxoCleanup;
+  private static intervalRef?: NodeJS.Timeout;
+
+  private constructor() {}
+
+  static start(): GaslessUtxoCleanup {
+    if (!GaslessUtxoCleanup.instance) {
+      GaslessUtxoCleanup.instance = new GaslessUtxoCleanup();
+
+      GaslessUtxoCleanup.intervalRef = setInterval(async () => {
+        const db = await MongoDatabase.connect();
+        const utxos = gaslessUtxosCollection(
+          db.getCollection<GaslessUtxo>(COLLECTION_GASLESS_UTXOS)
+        );
+        const released = await utxos.releaseExpired();
+        if (released > 0) {
+          console.log(
+            `[GASLESS_UTXO_CLEANUP]: Released ${released} expired reservation(s).`
+          );
+        }
+      }, CLEANUP_INTERVAL_MS);
+    }
+
+    return GaslessUtxoCleanup.instance;
+  }
+
+  static stop(): void {
+    if (GaslessUtxoCleanup.intervalRef) {
+      clearInterval(GaslessUtxoCleanup.intervalRef);
+      GaslessUtxoCleanup.intervalRef = undefined;
+      console.log("[GASLESS_UTXO_CLEANUP]: Stopped.");
+    }
+  }
+}

packages/worker/src/queues/gaslessUtxos/index.ts

@@ -0,0 +1,20 @@
+import { Collection } from 'mongodb';
+import { GaslessUtxo, type ReserveUtxoOptions, GaslessUtxoStats } from './types';
+import { findAvailable } from './utils/findAvailable';
+import { reserve } from './utils/reserve';
+import { release } from './utils/release';
+import { markSpent } from './utils/markSpent';
+import { getStats } from './utils/getStats';
+import { releaseExpired } from './utils/releaseExpired';
+
+export const gaslessUtxosCollection = (collection: Collection<GaslessUtxo>) => ({
+  findAvailable: (): Promise<GaslessUtxo[]> => findAvailable(collection),
+  reserve: (options: ReserveUtxoOptions): Promise<GaslessUtxo | null> => reserve(collection, options),
+  release: (utxoId: string): Promise<GaslessUtxo | null> => release(collection, utxoId),
+  markSpent: (utxoId: string, spentTxHash: string): Promise<GaslessUtxo | null> => markSpent(collection, utxoId, spentTxHash),
+  getStats: (): Promise<GaslessUtxoStats> => getStats(collection),
+  releaseExpired: (): Promise<number> => releaseExpired(collection),
+});
+
+export * from './types';
+export * from './constants';

packages/worker/src/queues/gaslessUtxos/types.ts

@@ -0,0 +1,26 @@
+import { ObjectId } from "mongodb";
+
+export interface GaslessUtxo {
+  _id?: ObjectId;
+  utxoId: string;
+  txId: string;
+  outputIndex: number;
+  amount: string;
+  status: "available" | "reserved" | "spent";
+  reservedAt?: Date;
+  reservedBy?: string;
+  spentTxHash?: string;
+  createdAt: Date;
+}
+
+export interface GaslessUtxoStats {
+  available: number;
+  reserved: number;
+  spent: number;
+  total: number;
+}
+
+export interface ReserveUtxoOptions {
+  reservedBy: string;
+  ttlSeconds?: number;
+}

packages/worker/src/queues/gaslessUtxos/utils/findAvailable.ts

@@ -0,0 +1,8 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo } from "../types";
+
+export const findAvailable = async (
+  collection: Collection<GaslessUtxo>
+): Promise<GaslessUtxo[]> => {
+  return collection.find({ status: "available" }).toArray();
+};

packages/worker/src/queues/gaslessUtxos/utils/getStats.ts

@@ -0,0 +1,27 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo, GaslessUtxoStats } from "../types";
+
+export const getStats = async (
+  collection: Collection<GaslessUtxo>
+): Promise<GaslessUtxoStats> => {
+  const rows = await collection
+    .aggregate<{ _id: string; count: number }>([
+      { $group: { _id: "$status", count: { $sum: 1 } } },
+    ])
+    .toArray();
+
+  const stats: GaslessUtxoStats = {
+    available: 0,
+    reserved: 0,
+    spent: 0,
+    total: 0,
+  };
+
+  for (const row of rows) {
+    const key = row._id as keyof Omit<GaslessUtxoStats, "total">;
+    if (key in stats) stats[key] = row.count;
+    stats.total += row.count;
+  }
+
+  return stats;
+};

packages/worker/src/queues/gaslessUtxos/utils/markSpent.ts

@@ -0,0 +1,17 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo } from "../types";
+
+export const markSpent = async (
+  collection: Collection<GaslessUtxo>,
+  utxoId: string,
+  spentTxHash: string
+): Promise<GaslessUtxo | null> => {
+  return collection.findOneAndUpdate(
+    { utxoId, status: "reserved" },
+    {
+      $set: { status: "spent", spentTxHash },
+      $unset: { reservedBy: "", reservedAt: "" },
+    },
+    { returnDocument: "after" }
+  );
+};

packages/worker/src/queues/gaslessUtxos/utils/release.ts

@@ -0,0 +1,16 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo } from "../types";
+
+export const release = async (
+  collection: Collection<GaslessUtxo>,
+  utxoId: string
+): Promise<GaslessUtxo | null> => {
+  return collection.findOneAndUpdate(
+    { utxoId, status: "reserved" },
+    {
+      $set: { status: "available" },
+      $unset: { reservedBy: "", reservedAt: "" },
+    },
+    { returnDocument: "after" }
+  );
+};

packages/worker/src/queues/gaslessUtxos/utils/releaseExpired.ts

@@ -0,0 +1,19 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo } from "@/queues/gaslessUtxos";
+import { RESERVE_TTLS_MS } from "@/queues/gaslessUtxos";
+
+export const releaseExpired = async (
+  collection: Collection<GaslessUtxo>
+): Promise<number> => {
+  const expiredBefore = new Date(Date.now() - RESERVE_TTLS_MS);
+
+  const result = await collection.updateMany(
+    { status: "reserved", reservedAt: { $lte: expiredBefore } },
+    {
+      $set: { status: "available" },
+      $unset: { reservedBy: "", reservedAt: "" },
+    }
+  );
+
+  return result.modifiedCount;
+};

packages/worker/src/queues/gaslessUtxos/utils/reserve.ts

@@ -0,0 +1,22 @@
+import { Collection } from "mongodb";
+import { GaslessUtxo, ReserveUtxoOptions } from "../types";
+import { DEFAULT_TTL_SECONDS } from "@/queues/gaslessUtxos/constants";
+
+export const reserve = async (
+  collection: Collection<GaslessUtxo>,
+  options: ReserveUtxoOptions
+): Promise<GaslessUtxo | null> => {
+  const { reservedBy, ttlSeconds = DEFAULT_TTL_SECONDS } = options;
+
+  return collection.findOneAndUpdate(
+    { status: "available" },
+    {
+      $set: {
+        status: "reserved",
+        reservedBy,
+        reservedAt: new Date(),
+      },
+    },
+    { returnDocument: "after" }
+  );
+};