Merge pull request #927 from Junirezz/fix/847-channels-integrate-payment-channels-with-renewal-executor

[#847] [P2] Channels: Integrate payment channels with renewal executor

Author: Calebux

backend/src/index.ts

@@ -68,6 +68,7 @@ import { adminAuth } from './middleware/admin';
 import { createAdminLimiter, RateLimiterFactory } from './middleware/rate-limit-factory';
 import { scheduleAutoResume } from './jobs/auto-resume';
 import { startSettlementBatchJob } from './jobs/settlement-batch-job';
+import { startChannelSettlementJob } from './jobs/channel-settlement-job';
 import { startJobAlertMonitor, stopJobAlertMonitor } from './jobs/job-alert-monitor';
 import giftCardLedgerRoutes from './routes/gift-card-ledger';
 import notificationDeadLetterRoutes from './routes/notification-dead-letter';
@@ -487,6 +488,7 @@ const server = app.listen(PORT, async () => {
 
   scheduleAutoResume();
   startSettlementBatchJob();
+  startChannelSettlementJob();
   startJobAlertMonitor();
 
   telegramCommandService.init();

backend/src/jobs/channel-settlement-job.ts

@@ -0,0 +1,60 @@
+import cron from 'node-cron';
+import logger from '../config/logger';
+import { runWithCorrelationId } from '../middleware/requestContext';
+import { channelStateService } from '../services/channel-state';
+import { paymentChannelService } from '../services/payment-channel-service';
+import { settlementBatcher } from '../services/settlement-batcher';
+
+/**
+ * Periodically settles accumulated channel balances on-chain per user schedule
+ * (monthly or quarterly). Runs daily at 02:00 UTC.
+ */
+export function startChannelSettlementJob(): void {
+  cron.schedule('0 2 * * *', () =>
+    runWithCorrelationId('cron:channel-settlement', async (cid) => {
+      if (process.env.PAYMENT_CHANNELS_ENABLED !== 'true') return;
+
+      try {
+        const due = await channelStateService.getChannelsDueForSettlement();
+        if (due.length === 0) return;
+
+        for (const candidate of due) {
+          try {
+            await settlementBatcher.enqueue({
+              userId: candidate.userId,
+              subscriptionId: candidate.channelId,
+              amount: candidate.executorBalance,
+              settlementType: 'channel_close',
+              payload: { channelId: candidate.channelId },
+            });
+
+            await paymentChannelService.initiateClose(
+              candidate.userId,
+              candidate.channelId,
+            );
+            await channelStateService.markChannelSettled(
+              candidate.channelId,
+              candidate.userId,
+            );
+
+            logger.info('Channel scheduled for settlement', {
+              correlationId: cid,
+              channelId: candidate.channelId,
+              amount: candidate.executorBalance,
+            });
+          } catch (err) {
+            logger.error('Channel settlement failed', {
+              correlationId: cid,
+              channelId: candidate.channelId,
+              error: err instanceof Error ? err.message : String(err),
+            });
+          }
+        }
+      } catch (error) {
+        logger.error('Channel settlement job failed', { correlationId: cid, error });
+      }
+    }),
+  );
+
+  logger.info('Channel settlement cron job scheduled (daily 02:00 UTC)');
+}

backend/src/services/channel-state.ts

@@ -0,0 +1,151 @@
+import { supabase } from '../config/database';
+import logger from '../config/logger';
+import {
+  paymentChannelService,
+  type PaymentChannelRecord,
+} from './payment-channel-service';
+
+export type SettlementSchedule = 'monthly' | 'quarterly';
+
+export interface ChannelPaymentLog {
+  channelId: string;
+  userId: string;
+  subscriptionId: string;
+  amount: number;
+  sequenceNumber: number;
+}
+
+export interface ChannelSettlementCandidate {
+  channelId: string;
+  userId: string;
+  executorBalance: number;
+}
+
+export class ChannelStateService {
+  /**
+   * Returns the newest active channel with sufficient off-chain balance.
+   */
+  async findPayableChannel(
+    userId: string,
+    amount: number,
+  ): Promise<PaymentChannelRecord | null> {
+    const channels = await paymentChannelService.listChannels(userId);
+    for (const channel of channels) {
+      if (channel.state !== 'active') continue;
+      const balance =
+        channel.channelState?.userBalance ?? Number.parseFloat(channel.balance);
+      if (balance >= amount) return channel;
+    }
+    return null;
+  }
+
+  /**
+   * Applies an off-chain state update and records the payment locally (not on-chain).
+   */
+  async applyRenewalPayment(
+    channelId: string,
+    userId: string,
+    subscriptionId: string,
+    amount: number,
+  ): Promise<PaymentChannelRecord> {
+    const updated = await paymentChannelService.applyOffChainRenewal(
+      channelId,
+      userId,
+      amount,
+    );
+
+    await this.logChannelPayment({
+      channelId,
+      userId,
+      subscriptionId,
+      amount,
+      sequenceNumber: updated.channelState?.sequenceNumber ?? 0,
+    });
+
+    return updated;
+  }
+
+  async logChannelPayment(payment: ChannelPaymentLog): Promise<void> {
+    const { error } = await supabase.from('channel_payments').insert({
+      channel_id: payment.channelId,
+      user_id: payment.userId,
+      subscription_id: payment.subscriptionId,
+      amount: payment.amount,
+      sequence_number: payment.sequenceNumber,
+      created_at: new Date().toISOString(),
+    });
+
+    if (error) {
+      logger.warn('Failed to log channel payment', {
+        channelId: payment.channelId,
+        error: error.message,
+      });
+    }
+  }
+
+  async getSettlementSchedule(userId: string): Promise<SettlementSchedule> {
+    const { data } = await supabase
+      .from('profiles')
+      .select('channel_settlement_schedule')
+      .eq('id', userId)
+      .maybeSingle();
+
+    return data?.channel_settlement_schedule === 'quarterly' ? 'quarterly' : 'monthly';
+  }
+
+  /**
+   * Active channels whose executor-side balance should be settled on-chain.
+   */
+  async getChannelsDueForSettlement(): Promise<ChannelSettlementCandidate[]> {
+    const { data: channels, error } = await supabase
+      .from('payment_channels')
+      .select('id, user_id, channel_state, last_settlement_at, state')
+      .eq('state', 'active');
+
+    if (error) throw error;
+
+    const due: ChannelSettlementCandidate[] = [];
+    const now = Date.now();
+
+    for (const row of channels ?? []) {
+      const state = row.channel_state as { executorBalance?: number } | null;
+      const executorBalance = state?.executorBalance ?? 0;
+      if (executorBalance <= 0) continue;
+
+      const schedule = await this.getSettlementSchedule(row.user_id as string);
+      const intervalMs =
+        schedule === 'quarterly'
+          ? 90 * 24 * 60 * 60 * 1000
+          : 30 * 24 * 60 * 60 * 1000;
+
+      const lastSettlement = row.last_settlement_at
+        ? new Date(row.last_settlement_at as string).getTime()
+        : 0;
+
+      if (now - lastSettlement >= intervalMs) {
+        due.push({
+          channelId: row.id as string,
+          userId: row.user_id as string,
+          executorBalance,
+        });
+      }
+    }
+
+    return due;
+  }
+
+  async markChannelSettled(channelId: string, userId: string): Promise<void> {
+    const { error } = await supabase
+      .from('payment_channels')
+      .update({
+        last_settlement_at: new Date().toISOString(),
+        updated_at: new Date().toISOString(),
+      })
+      .eq('id', channelId)
+      .eq('user_id', userId);
+
+    if (error) throw error;
+  }
+}
+
+export const channelStateService = new ChannelStateService();

backend/src/services/renewal-executor.ts

@@ -2,7 +2,7 @@ import logger from '../config/logger';
 import { supabase } from '../config/database';
 import { blockchainService } from './blockchain-service';
 import { webhookService } from './webhook-service';
-import { paymentChannelService } from './payment-channel-service';
+import { channelStateService } from './channel-state';
 import { settlementBatcher } from './settlement-batcher';
 import { addMonths, addQuarters, addYears } from 'date-fns';
 import { deriveEphemeralStealthAddress } from '@syncro/shared/crypto';
@@ -240,22 +240,26 @@ export class RenewalExecutor {
       return { used: false };
     }
 
-    const { data: channel } = await supabase
-      .from('payment_channels')
-      .select('id')
-      .eq('user_id', userId)
-      .eq('state', 'active')
-      .order('created_at', { ascending: false })
-      .limit(1)
-      .maybeSingle();
-
-    if (!channel) return { used: false };
-
     try {
-      await paymentChannelService.applyOffChainRenewal(channel.id, userId, amount);
-      logger.info('Off-chain channel renewal applied', { channelId: channel.id, subscriptionId });
+      const channel = await channelStateService.findPayableChannel(userId, amount);
+      if (!channel) return { used: false };
+
+      await channelStateService.applyRenewalPayment(
+        channel.id,
+        userId,
+        subscriptionId,
+        amount,
+      );
+      logger.info('Off-chain channel renewal applied', {
+        channelId: channel.id,
+        subscriptionId,
+      });
       return { used: true };
-    } catch {
+    } catch (err) {
+      logger.warn('Channel renewal failed, falling back to on-chain', {
+        subscriptionId,
+        error: err instanceof Error ? err.message : String(err),
+      });
       return { used: false };
     }
   }

backend/tests/channel-state.test.ts

@@ -0,0 +1,112 @@
+jest.mock('../src/config/database', () => ({
+  supabase: { from: jest.fn() },
+}));
+
+jest.mock('../src/config/logger', () => ({
+  default: { info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn() },
+  __esModule: true,
+}));
+
+jest.mock('../src/services/payment-channel-service', () => ({
+  paymentChannelService: {
+    listChannels: jest.fn(),
+    applyOffChainRenewal: jest.fn(),
+  },
+}));
+
+import { ChannelStateService } from '../src/services/channel-state';
+import { supabase } from '../src/config/database';
+import { paymentChannelService } from '../src/services/payment-channel-service';
+
+describe('ChannelStateService', () => {
+  let service: ChannelStateService;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    service = new ChannelStateService();
+  });
+
+  describe('findPayableChannel', () => {
+    it('returns active channel with sufficient balance', async () => {
+      (paymentChannelService.listChannels as jest.Mock).mockResolvedValue([
+        {
+          id: 'ch-1',
+          state: 'active',
+          balance: '50',
+          channelState: { userBalance: 50, executorBalance: 0, sequenceNumber: 0, totalDeposited: 50 },
+        },
+        {
+          id: 'ch-2',
+          state: 'active',
+          balance: '5',
+          channelState: { userBalance: 5, executorBalance: 0, sequenceNumber: 0, totalDeposited: 5 },
+        },
+      ]);
+
+      const channel = await service.findPayableChannel('user-1', 10);
+      expect(channel?.id).toBe('ch-1');
+    });
+
+    it('returns null when no channel has enough balance', async () => {
+      (paymentChannelService.listChannels as jest.Mock).mockResolvedValue([
+        {
+          id: 'ch-1',
+          state: 'active',
+          balance: '5',
+          channelState: { userBalance: 5, executorBalance: 0, sequenceNumber: 0, totalDeposited: 5 },
+        },
+      ]);
+
+      const channel = await service.findPayableChannel('user-1', 10);
+      expect(channel).toBeNull();
+    });
+  });
+
+  describe('applyRenewalPayment', () => {
+    it('updates channel state and logs payment locally', async () => {
+      const updatedChannel = {
+        id: 'ch-1',
+        state: 'active',
+        balance: '40',
+        channelState: { userBalance: 40, executorBalance: 10, sequenceNumber: 1, totalDeposited: 50 },
+      };
+
+      (paymentChannelService.applyOffChainRenewal as jest.Mock).mockResolvedValue(updatedChannel);
+      (supabase.from as jest.Mock).mockReturnValue({
+        insert: jest.fn().mockResolvedValue({ error: null }),
+      });
+
+      const result = await service.applyRenewalPayment('ch-1', 'user-1', 'sub-1', 10);
+
+      expect(paymentChannelService.applyOffChainRenewal).toHaveBeenCalledWith('ch-1', 'user-1', 10);
+      expect(supabase.from).toHaveBeenCalledWith('channel_payments');
+      expect(result.channelState?.sequenceNumber).toBe(1);
+    });
+  });
+
+  describe('getSettlementSchedule', () => {
+    it('defaults to monthly', async () => {
+      (supabase.from as jest.Mock).mockReturnValue({
+        select: jest.fn().mockReturnThis(),
+        eq: jest.fn().mockReturnThis(),
+        maybeSingle: jest.fn().mockResolvedValue({ data: null }),
+      });
+
+      const schedule = await service.getSettlementSchedule('user-1');
+      expect(schedule).toBe('monthly');
+    });
+
+    it('returns quarterly when configured', async () => {
+      (supabase.from as jest.Mock).mockReturnValue({
+        select: jest.fn().mockReturnThis(),
+        eq: jest.fn().mockReturnThis(),
+        maybeSingle: jest.fn().mockResolvedValue({
+          data: { channel_settlement_schedule: 'quarterly' },
+        }),
+      });
+
+      const schedule = await service.getSettlementSchedule('user-1');
+      expect(schedule).toBe('quarterly');
+    });
+  });
+});