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 } 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 Promise>>(); // Pub/sub unsubscribe handles per ticket (one local subscription per ticket). const channelUnsubs = new Map void>(); // Store for active background checkers (ticketId -> intervalId) const activeCheckers = new Map(); /** 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(); /** 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; const lockToken = await getLock().acquire(`checker:${ticketId}`, expiryMs); if (!lockToken) { // Another instance is already polling this ticket. return; } 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( (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( (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) for (const ticket of ticketsToConfirm) { await (db as any) .update(tickets) .set({ status: 'confirmed' }) .where(and(eq((tickets as any).id, ticket.id), eq((tickets as any).status, 'pending'))); 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})`); } // Get primary payment for sending receipt const payment = await dbGet( (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( (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( (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); // Clean up on disconnect stream.onAbort(() => { 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); } } } }); // Keep the stream open while (true) { await stream.sleep(30000); } }); }); /** * 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( (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( (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( (db as any).select().from(tickets).where(eq((tickets as any).bookingId, ticket.bookingId)) ); } const groupPayments = await dbAll( (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( (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( (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( (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;