Spanglish/backend/src/routes/lnbits.ts

import { Hono } from 'hono';
import { streamSSE } from 'hono/streaming';
import { db, dbGet, dbAll, tickets, payments, events } from '../db/index.js';
import { eq, and, inArray } from 'drizzle-orm';
import { getNow, toDbDate } from '../lib/utils.js';
import {
  verifyWebhookPayment,
  getPaymentStatus,
  createInvoice,
  isLNbitsConfigured,
  LNBITS_INVOICE_EXPIRY_SECONDS,
} from '../lib/lnbits.js';
import emailService from '../lib/email.js';
import { getPubSub } from '../lib/stores/pubsub.js';
import { getLock, LockUnavailableError } from '../lib/stores/lock.js';

const lnbitsRouter = new Hono();

// Local SSE connections owned by THIS process (ticketId -> Set of response writers).
// Cross-instance delivery is handled by pub/sub: see paymentChannel below.
const activeConnections = new Map<string, Set<(data: any) => Promise<void>>>();

// Pub/sub unsubscribe handles per ticket (one local subscription per ticket).
const channelUnsubs = new Map<string, () => void>();

// Store for active background checkers (ticketId -> intervalId)
const activeCheckers = new Map<string, NodeJS.Timeout>();

/** Pub/sub channel that carries payment events for a ticket. */
function paymentChannel(ticketId: string): string {
  return `payment:${ticketId}`;
}

/**
 * LNbits webhook payload structure
 */
interface LNbitsWebhookPayload {
  payment_hash: string;
  payment_request?: string;
  amount: number;
  memo?: string;
  status: string;
  preimage?: string;
  extra?: {
    ticketId?: string;
    eventId?: string;
    [key: string]: any;
  };
}

/**
 * Notify every client for a ticket across all instances.
 *
 * Publishes to the ticket's pub/sub channel. In single-instance / in-memory
 * mode this is an in-process broadcast; with Redis it reaches whichever
 * instance(s) actually hold the SSE socket(s) for this ticket.
 */
async function notifyClients(ticketId: string, data: any) {
  await getPubSub().publish(paymentChannel(ticketId), data);
}

/**
 * Deliver an event to the SSE sockets held by THIS process for a ticket.
 * Invoked by the pub/sub subscription handler.
 */
async function deliverLocal(ticketId: string, data: any) {
  const connections = activeConnections.get(ticketId);
  if (connections) {
    await Promise.all(
      Array.from(connections).map(async (send) => {
        try {
          await send(data);
        } catch (e) {
          // Connection might be closed
        }
      })
    );
  }
}

// Distributed lock tokens for the per-ticket poller (ticketId -> token).
const checkerLockTokens = new Map<string, string>();

/** Release the per-ticket poller lock if this process holds it. */
function releaseCheckerLock(ticketId: string) {
  const token = checkerLockTokens.get(ticketId);
  if (token) {
    checkerLockTokens.delete(ticketId);
    void getLock().release(`checker:${ticketId}`, token);
  }
}

/**
 * Start background payment checking for a ticket.
 *
 * Only one instance should poll LNbits per ticket, so we take a distributed
 * lock for the lifetime of the poll. Other instances skip polling and instead
 * receive the result via pub/sub. With no Redis configured the lock is a local
 * no-op and behavior matches the original single-instance polling.
 */
async function startBackgroundChecker(ticketId: string, paymentHash: string, expirySeconds: number = 900) {
  // Don't start if already checking on this instance
  if (activeCheckers.has(ticketId)) {
    return;
  }

  const expiryMs = expirySeconds * 1000;

  let lockToken: string | null = null;
  let lockUnavailable = false;
  try {
    lockToken = await getLock().acquire(`checker:${ticketId}`, expiryMs);
  } catch (err) {
    if (!(err instanceof LockUnavailableError)) throw err;
    // Fail open: in webhook-less deployments this poller is the only way a
    // Lightning payment gets confirmed, so a Redis outage must not stop it.
    // Duplicate pollers across replicas are harmless because
    // handlePaymentComplete only acts on rows it actually transitions.
    console.warn(`[lnbits] lock backend unavailable, polling ticket ${ticketId} without lock:`, err.message);
    lockUnavailable = true;
  }
  if (!lockToken && !lockUnavailable) {
    // Another instance is already polling this ticket.
    return;
  }
  if (lockToken) {
    checkerLockTokens.set(ticketId, lockToken);
  }

  const startTime = Date.now();
  let checkCount = 0;

  console.log(`Starting background checker for ticket ${ticketId}, expires in ${expirySeconds}s`);

  const checkInterval = setInterval(async () => {
    checkCount++;
    const elapsed = Date.now() - startTime;

    // Stop if expired
    if (elapsed >= expiryMs) {
      console.log(`Invoice expired for ticket ${ticketId}`);
      clearInterval(checkInterval);
      activeCheckers.delete(ticketId);
      releaseCheckerLock(ticketId);
      await notifyClients(ticketId, { type: 'expired', ticketId });
      return;
    }

    try {
      const status = await getPaymentStatus(paymentHash);
      
      if (status?.paid) {
        console.log(`Payment confirmed for ticket ${ticketId} (check #${checkCount})`);
        clearInterval(checkInterval);
        activeCheckers.delete(ticketId);
        releaseCheckerLock(ticketId);
        
        await handlePaymentComplete(ticketId, paymentHash);
        await notifyClients(ticketId, { type: 'paid', ticketId, paymentHash });
      }
    } catch (error) {
      console.error(`Error checking payment for ticket ${ticketId}:`, error);
    }
  }, 3000); // Check every 3 seconds

  activeCheckers.set(ticketId, checkInterval);
}

/**
 * Stop background checker for a ticket
 */
function stopBackgroundChecker(ticketId: string) {
  const interval = activeCheckers.get(ticketId);
  if (interval) {
    clearInterval(interval);
    activeCheckers.delete(ticketId);
    releaseCheckerLock(ticketId);
  }
}

/**
 * LNbits webhook endpoint
 * Called by LNbits when a payment is received
 */
lnbitsRouter.post('/webhook', async (c) => {
  try {
    // Optional shared-secret gate: if LNBITS_WEBHOOK_SECRET is configured, the
    // webhook URL must carry a matching ?token=... (set when the invoice is created).
    const webhookSecret = process.env.LNBITS_WEBHOOK_SECRET || '';
    if (webhookSecret) {
      const provided = c.req.query('token') || c.req.header('x-webhook-secret') || '';
      if (provided !== webhookSecret) {
        console.warn('LNbits webhook rejected: invalid or missing secret');
        return c.json({ received: true, processed: false }, 401);
      }
    }

    const payload: LNbitsWebhookPayload = await c.req.json();
    
    // Log identifiers only (no full payload / PII)
    console.log('LNbits webhook received:', {
      paymentHash: payload.payment_hash,
      status: payload.status,
    });

    // Verify the payment is actually complete by checking with LNbits
    const isVerified = await verifyWebhookPayment(payload.payment_hash);
    
    if (!isVerified) {
      console.warn('LNbits webhook payment not verified:', payload.payment_hash);
      return c.json({ received: true, processed: false }, 200);
    }

    const ticketId = payload.extra?.ticketId;
    
    if (!ticketId) {
      console.error('No ticketId in LNbits webhook extra data');
      return c.json({ received: true, processed: false }, 200);
    }

    // CRITICAL: bind the paid hash to this ticket's own invoice. Without this, a
    // valid paid hash from any other invoice could be replayed with an arbitrary
    // ticketId to confirm tickets for free.
    const ticketPayment = await dbGet<any>(
      (db as any).select().from(payments).where(eq((payments as any).ticketId, ticketId))
    );
    if (!ticketPayment || ticketPayment.reference !== payload.payment_hash) {
      console.warn('LNbits webhook rejected: payment hash does not match the ticket invoice', {
        ticketId,
        paymentHash: payload.payment_hash,
      });
      return c.json({ received: true, processed: false }, 200);
    }

    // Stop background checker since webhook confirmed payment
    stopBackgroundChecker(ticketId);

    await handlePaymentComplete(ticketId, payload.payment_hash);
    
    // Notify connected clients via SSE
    await notifyClients(ticketId, { type: 'paid', ticketId, paymentHash: payload.payment_hash });
    
    return c.json({ received: true, processed: true }, 200);
  } catch (error) {
    console.error('LNbits webhook error:', error);
    return c.json({ error: 'Webhook processing failed' }, 500);
  }
});

/**
 * Handle successful payment
 * Supports multi-ticket bookings - confirms all tickets in the booking
 */
async function handlePaymentComplete(ticketId: string, paymentHash: string) {
  const now = getNow();
  
  // Get the ticket to check for booking ID
  const existingTicket = await dbGet<any>(
    (db as any).select().from(tickets).where(eq((tickets as any).id, ticketId))
  );
  
  if (!existingTicket) {
    console.error(`Ticket ${ticketId} not found for payment confirmation`);
    return;
  }
  
  if (existingTicket.status === 'confirmed') {
    console.log(`Ticket ${ticketId} already confirmed, skipping update`);
    return;
  }
  
  // Get all tickets in this booking (if multi-ticket)
  let ticketsToConfirm: any[] = [existingTicket];
  
  if (existingTicket.bookingId) {
    // This is a multi-ticket booking - get all tickets with same bookingId
    ticketsToConfirm = await dbAll(
      (db as any)
        .select()
        .from(tickets)
        .where(eq((tickets as any).bookingId, existingTicket.bookingId))
    );
    console.log(`Multi-ticket booking detected: ${ticketsToConfirm.length} tickets to confirm`);
  }
  
  // Confirm all tickets in the booking (idempotent: only flip pending -> confirmed)
  let transitioned = 0;
  for (const ticket of ticketsToConfirm) {
    const result: any = await (db as any)
      .update(tickets)
      .set({ status: 'confirmed', paymentStatus: 'paid' })
      .where(and(eq((tickets as any).id, ticket.id), eq((tickets as any).status, 'pending')));
    transitioned += result?.changes ?? result?.rowCount ?? 0;

    await (db as any)
      .update(payments)
      .set({
        status: 'paid',
        reference: paymentHash,
        paidAt: now,
        updatedAt: now,
      })
      .where(and(eq((payments as any).ticketId, ticket.id), eq((payments as any).status, 'pending')));

    console.log(`Ticket ${ticket.id} confirmed via Lightning payment (hash: ${paymentHash})`);
  }

  // Only the caller that actually flipped rows sends the emails. The webhook
  // and the background poller (and pollers on multiple replicas during a Redis
  // outage) can all land here; without this guard they would each email.
  if (transitioned === 0) {
    console.log(`Ticket ${ticketId} was already confirmed by a concurrent caller, skipping emails`);
    return;
  }

  // Get primary payment for sending receipt
  const payment = await dbGet<any>(
    (db as any)
      .select()
      .from(payments)
      .where(eq((payments as any).ticketId, ticketId))
  );

  // Send confirmation emails asynchronously
  // For multi-ticket bookings, send email with all ticket info
  Promise.all([
    emailService.sendBookingConfirmation(ticketId),
    payment ? emailService.sendPaymentReceipt(payment.id) : Promise.resolve(),
  ]).catch(err => {
    console.error('[Email] Failed to send confirmation emails:', err);
  });
}

/**
 * SSE endpoint for real-time payment status updates
 * Frontend connects here to receive instant payment notifications
 */
lnbitsRouter.get('/stream/:ticketId', async (c) => {
  const ticketId = c.req.param('ticketId');
  
  // Verify ticket exists
  const ticket = await dbGet<any>(
    (db as any).select().from(tickets).where(eq((tickets as any).id, ticketId))
  );
  
  if (!ticket) {
    return c.json({ error: 'Ticket not found' }, 404);
  }
  
  // Get payment to start background checker
  const payment = await dbGet<any>(
    (db as any).select().from(payments).where(eq((payments as any).ticketId, ticketId))
  );
  
  // Start background checker if not already running (only while still pending)
  if (ticket.status !== 'confirmed' && payment?.reference && !activeCheckers.has(ticketId)) {
    await startBackgroundChecker(ticketId, payment.reference, 900); // 15 min expiry
  }
  
  // Prevent proxies/CDNs from buffering the event stream so events flush immediately.
  c.header('Cache-Control', 'no-cache, no-transform');
  c.header('X-Accel-Buffering', 'no');
  
  return streamSSE(c, async (stream) => {
    const sendEvent = async (data: any) => {
      await stream.writeSSE({ data: JSON.stringify(data), event: 'payment' });
    };
    
    // If already paid, notify over SSE and close (EventSource can parse this).
    if (ticket.status === 'confirmed') {
      await sendEvent({ type: 'already_paid', ticketId });
      return;
    }
    
    // Register this connection. The first local connection for a ticket also
    // subscribes to the ticket's pub/sub channel so events published by any
    // instance (webhook or background checker) are delivered to these sockets.
    if (!activeConnections.has(ticketId)) {
      activeConnections.set(ticketId, new Set());
      const unsub = await getPubSub().subscribe(paymentChannel(ticketId), (data) => {
        void deliverLocal(ticketId, data);
      });
      channelUnsubs.set(ticketId, unsub);
    }
    activeConnections.get(ticketId)!.add(sendEvent);
    
    // Send initial status
    await sendEvent({ type: 'connected', ticketId });
    
    // Keep connection alive with heartbeat
    const heartbeat = setInterval(async () => {
      try {
        await stream.writeSSE({ data: 'ping', event: 'heartbeat' });
      } catch (e) {
        clearInterval(heartbeat);
      }
    }, 15000);
    
    const cleanup = () => {
      clearInterval(heartbeat);
      const connections = activeConnections.get(ticketId);
      if (connections) {
        connections.delete(sendEvent);
        if (connections.size === 0) {
          activeConnections.delete(ticketId);
          // Drop the pub/sub subscription once no local sockets remain.
          const unsub = channelUnsubs.get(ticketId);
          if (unsub) {
            unsub();
            channelUnsubs.delete(ticketId);
          }
        }
      }
    };

    // Clean up on disconnect
    stream.onAbort(cleanup);

    // Keep the stream open, and every 15s fall back to reading the ticket
    // status from the DB. Pub/sub is best-effort: a message published while the
    // subscriber connection was reconnecting is lost, and without this check
    // the client would wait forever on a payment that already confirmed.
    try {
      while (true) {
        await stream.sleep(15000);
        const current = await dbGet<any>(
          (db as any).select().from(tickets).where(eq((tickets as any).id, ticketId))
        );
        if (current?.status === 'confirmed') {
          await sendEvent({ type: 'paid', ticketId });
          return;
        }
      }
    } finally {
      cleanup();
    }
  });
});

/**
 * Get a Lightning invoice for a ticket to pay or re-pay.
 *
 * Reuses the stored invoice if it still has more than 5 minutes of validity
 * left; otherwise generates a fresh one from LNbits. This is what lets a user
 * come back to an unpaid Lightning booking later (e.g. from "Pay now" on the
 * dashboard) instead of hitting a dead end.
 */
lnbitsRouter.post('/invoice/:ticketId', async (c) => {
  const ticketId = c.req.param('ticketId');

  const ticket = await dbGet<any>(
    (db as any).select().from(tickets).where(eq((tickets as any).id, ticketId))
  );
  if (!ticket) {
    return c.json({ error: 'Ticket not found' }, 404);
  }

  const payment = await dbGet<any>(
    (db as any).select().from(payments).where(eq((payments as any).ticketId, ticketId))
  );
  if (!payment) {
    return c.json({ error: 'Payment not found' }, 404);
  }

  if (payment.provider !== 'lightning') {
    return c.json({ error: 'This booking is not a Lightning payment' }, 400);
  }

  if (ticket.status === 'confirmed' || payment.status === 'paid') {
    return c.json({ alreadyPaid: true });
  }

  if (ticket.status !== 'pending' || payment.status !== 'pending') {
    return c.json({
      error: 'This booking is no longer active. Please make a new booking.',
    }, 400);
  }

  // Gather every ticket/payment in the booking group - a multi-ticket booking
  // shares a single Lightning invoice for the combined total.
  let groupTickets: any[] = [ticket];
  if (ticket.bookingId) {
    groupTickets = await dbAll<any>(
      (db as any).select().from(tickets).where(eq((tickets as any).bookingId, ticket.bookingId))
    );
  }
  const groupPayments = await dbAll<any>(
    (db as any).select().from(payments).where(inArray((payments as any).ticketId, groupTickets.map((t: any) => t.id)))
  );
  const invoiceHolder = groupPayments.find((p: any) => p.reference) || payment;
  const totalAmount = groupPayments.reduce((sum: number, p: any) => sum + Number(p.amount), 0);
  const currency = invoiceHolder.currency;

  const now = Date.now();
  const FIVE_MIN_MS = 5 * 60 * 1000;
  const expiresAtMs = invoiceHolder.lnbitsExpiresAt ? new Date(invoiceHolder.lnbitsExpiresAt).getTime() : 0;

  if (invoiceHolder.reference && invoiceHolder.lnbitsInvoice && expiresAtMs - now > FIVE_MIN_MS) {
    return c.json({
      invoice: {
        paymentHash: invoiceHolder.reference,
        paymentRequest: invoiceHolder.lnbitsInvoice,
        amount: invoiceHolder.lnbitsAmountSats || 0,
        fiatAmount: totalAmount,
        fiatCurrency: currency,
        expiresAt: invoiceHolder.lnbitsExpiresAt,
      },
      reused: true,
    });
  }

  // No usable stored invoice (never created, expired, or expiring soon) - get a fresh one.
  if (!isLNbitsConfigured()) {
    return c.json({ error: 'Bitcoin Lightning payments are not available at this time' }, 400);
  }

  const event = await dbGet<any>(
    (db as any).select().from(events).where(eq((events as any).id, ticket.eventId))
  );

  const apiUrl = process.env.API_URL || 'http://localhost:3001';
  const webhookSecret = process.env.LNBITS_WEBHOOK_SECRET || '';
  const webhookUrl = webhookSecret
    ? `${apiUrl}/api/lnbits/webhook?token=${encodeURIComponent(webhookSecret)}`
    : `${apiUrl}/api/lnbits/webhook`;
  const attendeeName = `${ticket.attendeeFirstName} ${ticket.attendeeLastName || ''}`.trim();

  try {
    const lnbitsInvoice = await createInvoice({
      amount: totalAmount,
      unit: currency,
      memo: `Spanglish: ${event?.title || 'Event'} - ${attendeeName}${groupTickets.length > 1 ? ` (${groupTickets.length} tickets)` : ''}`,
      webhookUrl,
      expiry: LNBITS_INVOICE_EXPIRY_SECONDS,
      extra: {
        ticketId: invoiceHolder.ticketId,
        bookingId: ticket.bookingId || null,
        ticketIds: groupTickets.map((t: any) => t.id),
        eventId: ticket.eventId,
        eventTitle: event?.title,
        attendeeName,
        attendeeEmail: ticket.attendeeEmail,
        ticketCount: groupTickets.length,
      },
    });

    const lnbitsExpiresAt = toDbDate(new Date(now + LNBITS_INVOICE_EXPIRY_SECONDS * 1000));

    await (db as any)
      .update(payments)
      .set({
        reference: lnbitsInvoice.paymentHash,
        lnbitsInvoice: lnbitsInvoice.paymentRequest,
        lnbitsExpiresAt,
        lnbitsAmountSats: lnbitsInvoice.amount,
        updatedAt: getNow(),
      })
      .where(eq((payments as any).id, invoiceHolder.id));

    return c.json({
      invoice: {
        paymentHash: lnbitsInvoice.paymentHash,
        paymentRequest: lnbitsInvoice.paymentRequest,
        amount: lnbitsInvoice.amount,
        fiatAmount: lnbitsInvoice.fiatAmount ?? totalAmount,
        fiatCurrency: lnbitsInvoice.fiatCurrency ?? currency,
        expiresAt: lnbitsExpiresAt,
      },
      reused: false,
    });
  } catch (error: any) {
    console.error('Failed to create Lightning invoice:', error);
    return c.json({
      error: `Failed to create Lightning invoice: ${error.message || 'Unknown error'}`,
    }, 500);
  }
});

/**
 * Get payment status for a ticket (fallback polling endpoint)
 */
lnbitsRouter.get('/status/:ticketId', async (c) => {
  const ticketId = c.req.param('ticketId');
  
  const ticket = await dbGet<any>(
    (db as any)
      .select()
      .from(tickets)
      .where(eq((tickets as any).id, ticketId))
  );
  
  if (!ticket) {
    return c.json({ error: 'Ticket not found' }, 404);
  }
  
  const payment = await dbGet<any>(
    (db as any)
      .select()
      .from(payments)
      .where(eq((payments as any).ticketId, ticketId))
  );
  
  return c.json({
    ticketStatus: ticket.status,
    paymentStatus: payment?.status || 'unknown',
    isPaid: ticket.status === 'confirmed' || payment?.status === 'paid',
  });
});

/**
 * Manual payment check endpoint
 */
lnbitsRouter.post('/check/:paymentHash', async (c) => {
  const paymentHash = c.req.param('paymentHash');
  
  try {
    const status = await getPaymentStatus(paymentHash);
    
    if (!status) {
      return c.json({ error: 'Payment not found' }, 404);
    }
    
    return c.json({
      paymentHash,
      paid: status.paid,
      status: status.status,
    });
  } catch (error) {
    console.error('Error checking payment:', error);
    return c.json({ error: 'Failed to check payment status' }, 500);
  }
});

export default lnbitsRouter;