feat(treasury): add automated fee sweep service with daily cron and t…

backend/.env.example

@@ -68,3 +68,23 @@ MAX_MARKET_POOL_XLM=100000
 MAX_PLATFORM_EXPOSURE_XLM=500000
 # Admin email to receive risk alert emails
 ADMIN_EMAIL=admin@boxmeout.app
+
+# ── Treasury Sweep Configuration ──────────────────────────────
+# Balance threshold (in XLM) that triggers an immediate sweep
+TREASURY_SWEEP_THRESHOLD_XLM=1000
+
+# Destination wallet address where swept Treasury funds are sent
+TREASURY_WALLET_ADDRESS=GXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
+
+# Stellar contract ID for the Treasury smart contract
+TREASURY_CONTRACT_ID=CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4
+
+# ── Treasury Sweep Configuration ──────────────────────────────
+# Balance threshold (in XLM) that triggers an immediate sweep
+TREASURY_SWEEP_THRESHOLD_XLM=1000
+
+# Destination wallet address where swept Treasury funds are sent
+TREASURY_WALLET_ADDRESS=GXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
+
+# Stellar contract ID for the Treasury smart contract
+TREASURY_CONTRACT_ID=CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4

backend/src/cron/treasurySweep.cron.ts

@@ -0,0 +1,71 @@
+import cron from 'node-cron';
+import { logger } from '../utils/logger';
+import { TreasuryService } from '../services/TreasuryService';
+
+let dailyCronTask: cron.ScheduledTask | null = null;
+let thresholdIntervalTask: NodeJS.Timeout | null = null;
+
+/**
+ * Register the Treasury sweep cron jobs
+ * 1. Daily sweep at 02:00 UTC via node-cron
+ * 2. Balance check every 10 minutes with threshold-based trigger
+ */
+export function registerTreasurySweepCrons(treasuryService: TreasuryService): void {
+  // Daily cron at 02:00 UTC (cron expression: minute hour day month day-of-week)
+  // 0 2 * * * = 02:00 every day
+  dailyCronTask = cron.schedule('0 2 * * *', async () => {
+    logger.info('Treasury sweep cron triggered: Daily scheduled sweep at 02:00 UTC');
+    await treasuryService.executeSweepCycle();
+  });
+
+  logger.info('Daily Treasury sweep cron registered: 0 2 * * * (02:00 UTC)');
+
+  // Threshold-based trigger: check balance every 10 minutes
+  thresholdIntervalTask = setInterval(async () => {
+    logger.debug('Checking Treasury balance for threshold-based trigger...');
+    try {
+      const balanceStroops = await treasuryService.getTreasuryBalance();
+      const thresholdStroops = (10_000_000 * 1000); // Default 1000 XLM in stroops
+
+      if (balanceStroops >= thresholdStroops) {
+        logger.info(
+          `Balance threshold exceeded: ${balanceStroops / 10_000_000} XLM >= 1000 XLM. Triggering sweep.`
+        );
+        await treasuryService.executeSweepCycle();
+      }
+    } catch (error) {
+      logger.error('Error during threshold check', { error });
+    }
+  }, 10 * 60 * 1000); // 10 minutes in milliseconds
+
+  logger.info('Treasury balance threshold check registered: every 10 minutes');
+}
+
+/**
+ * Gracefully stop the Treasury sweep crons
+ */
+export function stopTreasurySweepCrons(): void {
+  if (dailyCronTask) {
+    dailyCronTask.stop();
+    dailyCronTask.destroy();
+    logger.info('Daily Treasury sweep cron stopped');
+  }
+
+  if (thresholdIntervalTask) {
+    clearInterval(thresholdIntervalTask);
+    logger.info('Treasury threshold check interval stopped');
+  }
+}
+
+/**
+ * Get cron status for monitoring
+ */
+export function getTreasurySweepCronStatus(): {
+  dailyCronActive: boolean;
+  thresholdCheckActive: boolean;
+} {
+  return {
+    dailyCronActive: dailyCronTask !== null && dailyCronTask.status === 'running',
+    thresholdCheckActive: thresholdIntervalTask !== null,
+  };
+}

backend/src/db/migrations/0001_create_treasury_sweeps_table.sql

@@ -0,0 +1,30 @@
+-- Migration: Add treasury_sweeps table for audit trail
+-- Created: 2026-06-20
+-- Description: Create table to record all Treasury sweep events with full audit trail
+
+CREATE TABLE IF NOT EXISTS treasury_sweeps (
+  id SERIAL PRIMARY KEY,
+  swept_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  amount_xlm NUMERIC(20, 7) NOT NULL,
+  amount_stroops NUMERIC(20, 0) NOT NULL,
+  tx_hash VARCHAR(255) UNIQUE,
+  to_address VARCHAR(255) NOT NULL,
+  status VARCHAR(20) NOT NULL DEFAULT 'pending',
+  created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Create indexes for common queries
+CREATE INDEX idx_treasury_sweeps_swept_at ON treasury_sweeps(swept_at DESC);
+CREATE INDEX idx_treasury_sweeps_status ON treasury_sweeps(status);
+CREATE INDEX idx_treasury_sweeps_tx_hash ON treasury_sweeps(tx_hash);
+CREATE INDEX idx_treasury_sweeps_to_address ON treasury_sweeps(to_address);
+
+-- Add comments for documentation
+COMMENT ON TABLE treasury_sweeps IS 'Audit trail for Treasury fee sweep operations';
+COMMENT ON COLUMN treasury_sweeps.id IS 'Unique identifier for the sweep event';
+COMMENT ON COLUMN treasury_sweeps.swept_at IS 'Timestamp when the sweep was executed';
+COMMENT ON COLUMN treasury_sweeps.amount_xlm IS 'Amount swept in XLM';
+COMMENT ON COLUMN treasury_sweeps.amount_stroops IS 'Amount swept in stroops (on-chain unit)';
+COMMENT ON COLUMN treasury_sweeps.tx_hash IS 'Blockchain transaction hash (null if failed)';
+COMMENT ON COLUMN treasury_sweeps.to_address IS 'Destination wallet address';
+COMMENT ON COLUMN treasury_sweeps.status IS 'Status of the sweep: success or failed';

backend/src/db/schema/treasurySweeps.schema.ts

@@ -0,0 +1,32 @@
+import { pgTable, serial, timestamp, numeric, varchar, text } from 'drizzle-orm/pg-core';
+
+/**
+ * Treasury sweeps table schema
+ * Records all Treasury sweep events with full audit trail
+ */
+export const treasurySweeps = pgTable('treasury_sweeps', {
+  // Primary key
+  id: serial('id').primaryKey(),
+
+  // Timestamp of when the sweep was executed
+  swept_at: timestamp('swept_at', { withTimezone: true }).notNull().defaultNow(),
+
+  // Amount swept in XLM (decimal for precision)
+  amount_xlm: numeric('amount_xlm', { precision: 20, scale: 7 }).notNull(),
+
+  // Amount swept in stroops (for on-chain reference)
+  amount_stroops: numeric('amount_stroops', { precision: 20, scale: 0 }).notNull(),
+
+  // Transaction hash from the blockchain
+  tx_hash: varchar('tx_hash', { length: 255 }).unique(),
+
+  // Destination wallet address
+  to_address: varchar('to_address', { length: 255 }).notNull(),
+
+  // Status: 'success' if sweep completed, 'failed' if all retries exhausted
+  status: varchar('status', { length: 20 }).notNull().default('pending'),
+});
+
+// TypeScript type inference
+export type TreasurySweep = typeof treasurySweeps.$inferSelect;
+export type NewTreasurySweep = typeof treasurySweeps.$inferInsert;

backend/src/repositories/TreasurySweepRepository.ts

@@ -0,0 +1,137 @@
+import { db } from '../db/client';
+import { treasurySweeps } from '../db/schema';
+import { eq, and, desc } from 'drizzle-orm';
+import { logger } from '../utils/logger';
+
+export interface TreasurySweepRecord {
+  amount_xlm: number;
+  amount_stroops: number;
+  tx_hash: string | null;
+  to_address: string;
+  status: 'success' | 'failed';
+}
+
+export class TreasurySweepRepository {
+  /**
+   * Record a Treasury sweep event in the database
+   * Ensures idempotency: if a tx_hash already exists, it will not insert a duplicate row
+   */
+  async recordSweep(data: TreasurySweepRecord): Promise<void> {
+    try {
+      // Check for existing record with same tx_hash (idempotency check)
+      if (data.tx_hash) {
+        const existingRecord = await db
+          .select()
+          .from(treasurySweeps)
+          .where(eq(treasurySweeps.tx_hash, data.tx_hash))
+          .limit(1);
+
+        if (existingRecord.length > 0) {
+          logger.info(
+            `Treasury sweep with tx_hash ${data.tx_hash} already recorded. Skipping duplicate insertion.`
+          );
+          return;
+        }
+      }
+
+      // Insert the sweep record
+      const result = await db.insert(treasurySweeps).values({
+        swept_at: new Date(),
+        amount_xlm: data.amount_xlm,
+        amount_stroops: data.amount_stroops,
+        tx_hash: data.tx_hash,
+        to_address: data.to_address,
+        status: data.status,
+      });
+
+      logger.info(
+        `Treasury sweep recorded: status=${data.status}, amount=${data.amount_xlm} XLM, tx_hash=${data.tx_hash || 'N/A'}`
+      );
+    } catch (error) {
+      logger.error('Failed to record Treasury sweep', { error, data });
+      throw error;
+    }
+  }
+
+  /**
+   * Find the latest sweep record(s) by date
+   */
+  async findLatestBySweepDate(afterDate: Date): Promise<any | null> {
+    try {
+      const results = await db
+        .select()
+        .from(treasurySweeps)
+        .where(
+          and(
+            // @ts-ignore - Drizzle ORM date comparison
+            treasurySweeps.swept_at >= afterDate
+          )
+        )
+        .orderBy(desc(treasurySweeps.swept_at))
+        .limit(1);
+
+      return results.length > 0 ? results[0] : null;
+    } catch (error) {
+      logger.error('Failed to find latest Treasury sweep', { error });
+      throw error;
+    }
+  }
+
+  /**
+   * Get all sweep records with optional filtering
+   */
+  async getAllSweeps(
+    filters?: {
+      status?: 'success' | 'failed';
+      limit?: number;
+      offset?: number;
+    }
+  ): Promise<any[]> {
+    try {
+      let query = db.select().from(treasurySweeps);
+
+      if (filters?.status) {
+        query = query.where(eq(treasurySweeps.status, filters.status));
+      }
+
+      query = query.orderBy(desc(treasurySweeps.swept_at));
+
+      if (filters?.limit) {
+        query = query.limit(filters.limit);
+      }
+
+      if (filters?.offset) {
+        query = query.offset(filters.offset);
+      }
+
+      return query;
+    } catch (error) {
+      logger.error('Failed to retrieve Treasury sweeps', { error });
+      throw error;
+    }
+  }
+
+  /**
+   * Get sweep statistics
+   */
+  async getSweepStatistics(): Promise<{
+    totalSweeps: number;
+    successfulSweeps: number;
+    failedSweeps: number;
+    totalAmountXlm: number;
+  }> {
+    try {
+      const allSweeps = await db.select().from(treasurySweeps);
+
+      return {
+        totalSweeps: allSweeps.length,
+        successfulSweeps: allSweeps.filter((s) => s.status === 'success').length,
+        failedSweeps: allSweeps.filter((s) => s.status === 'failed').length,
+        totalAmountXlm: allSweeps.reduce((sum, s) => sum + s.amount_xlm, 0),
+      };
+    } catch (error) {
+      logger.error('Failed to calculate sweep statistics', { error });
+      throw error;
+    }
+  }
+}