Merge pull request #2 from nanopay/migrate-payments-notifier

Migrate payments notifier and refactor router

Author: anarkrypto

README.md

@@ -36,7 +36,6 @@ npm install
 ```bash
 wrangler queues create payment-listener-queue
 wrangler queues create payment-write-queue
-wrangler queues create payment-pusher-queue
 wrangler queues create payment-receiver-queue
 wrangler queues create payment-sender-queue
 wrangler queues create webhook-delivery-queue
@@ -73,11 +72,6 @@ wrangler secret put RPC_URL
 
 # A Nano node worker RPC url with PoW generation support (can be the same as RPC_URL)
 wrangler secret put WORKER_URLS
-
-# Your Pusher crendentials
-wrangler secret put PUSHER_APP_ID
-wrangler secret put PUSHER_KEY
-wrangler secret put PUSHER_SECRET
 ```
 
 ## Running locally

example.dev.vars

@@ -5,10 +5,6 @@ SUPABASE_SECRET_KEY="xxx.xxx.xx-xxx"
 
 NANO_WEBSOCKET_URL="wss://xx.xxxx.xx"
 
-PUSHER_APP_ID="xxxx"
-PUSHER_KEY="xxxx"
-PUSHER_SECRET="xxxx"
-
 HOT_WALLET_SEED="xxxxxxxx"
 REPRESENTATIVE="nano_3xxx"
 RPC_URLS=https://rpc.xxxx.xx

package-lock.json

@@ -11,20 +11,18 @@
 	"dependencies": {
 		"@supabase/supabase-js": "^2.8.0",
 		"bignumber.js": "^9.1.1",
-		"crypto-js": "^4.1.1",
 		"nanocurrency": "^2.5.0",
 		"path-to-regexp": "^7.1.0",
 		"uuid": "^9.0.1",
 		"zod": "^3.22.4"
 	},
 	"devDependencies": {
 		"@cloudflare/workers-types": "^4.20240718.0",
-		"@types/crypto-js": "^4.1.1",
 		"@types/uuid": "^9.0.8",
 		"typescript": "^5.5.2",
 		"wrangler": "3.65.1"
 	},
 	"resolutions": {
 		"@types/node": "18.15.3"
 	}
-}
+}

package.json

@@ -4,10 +4,12 @@ import { Invoice, MessageBody, Payment, Service, Webhook } from '../types';
 import { rawToNano } from '../utils';
 import { logger } from '../logger';
 import { INVOICE_MIN_AMOUNT } from '../constants';
+import { PaymentNotifier } from './payment-notifier';
 
 export class PaymentListener extends DurableObject<Env> {
 	private nanoWebsocket: NanoWebsocket;
 	private pendingInvoices: { id: string; expiresAt: string; payAddress: string; payments: Payment[] }[] = [];
+	private notifierNamespace: DurableObjectNamespace<PaymentNotifier>;
 
 	/**
 	 * The constructor is invoked once upon creation of the Durable Object, i.e. the first call to
@@ -19,6 +21,7 @@ export class PaymentListener extends DurableObject<Env> {
 	constructor(ctx: DurableObjectState, env: Env) {
 		super(ctx, env);
 		this.nanoWebsocket = new NanoWebsocket(env.NANO_WEBSOCKET_URL);
+		this.notifierNamespace = env.PAYMENT_NOTIFIER;
 	}
 
 	async listen(message: MessageBody) {
@@ -58,6 +61,8 @@ export class PaymentListener extends DurableObject<Env> {
 			payments: [],
 		});
 
+		await this.startPaymentNotifier(invoice.id);
+
 		await this.alarm();
 	}
 
@@ -101,23 +106,23 @@ export class PaymentListener extends DurableObject<Env> {
 
 		payments.push(newPayment);
 
-		this.paymentNotify(payments, invoice);
+		this.paymentNotify(newPayment, invoice.id);
 		this.paymentWrite(newPayment, invoice, service, webhooks);
 
 		const paid_total = payments.reduce((acc, payment) => {
 			return acc + payment.amount;
 		}, 0);
 
 		if (paid_total >= invoice.price) {
-			this.nanoWebsocket.unsubscribe(invoice.pay_address);
-			this.removePendingInvoice(invoice.id);
-
-			this.paymentReceiver(payments, invoice);
+			await this.removePendingInvoice(invoice.id, invoice.pay_address);
+			await this.paymentReceiver(payments, invoice);
 		}
 	}
 
-	private removePendingInvoice(invoiceId: string) {
+	private async removePendingInvoice(invoiceId: string, payAddress: string) {
+		this.nanoWebsocket.unsubscribe(payAddress);
 		this.pendingInvoices = this.pendingInvoices.filter((activeInvoice) => activeInvoice.id !== invoiceId);
+		await this.stopPaymentNotifier(invoiceId);
 	}
 
 	private async paymentWrite(payment: Payment, invoice: Invoice, service: Service, webhooks: Webhook[]) {
@@ -130,11 +135,28 @@ export class PaymentListener extends DurableObject<Env> {
 		});
 	}
 
-	private async paymentNotify(payments: Payment[], invoice: Invoice) {
-		// Send the payment to the worker to push to the channel
-		await this.env.PAYMENT_PUSHER_QUEUE.send({
-			invoice,
-			payments,
+	private async startPaymentNotifier(invoiceId: string) {
+		const notifierId = this.notifierNamespace.idFromName(invoiceId);
+		const paymentNotifier = this.notifierNamespace.get(notifierId);
+		await paymentNotifier.start();
+	}
+
+	private async stopPaymentNotifier(invoiceId: string) {
+		const notifierId = this.notifierNamespace.idFromName(invoiceId);
+		const paymentNotifier = this.notifierNamespace.get(notifierId);
+		await paymentNotifier.stop();
+	}
+
+	private async paymentNotify(payment: Payment, invoiceId: string) {
+		// Send the payment to the PaymentNotifier
+		const notifierId = this.notifierNamespace.idFromName(invoiceId);
+		const paymentNotifier = this.notifierNamespace.get(notifierId);
+		await paymentNotifier.notify({
+			from: payment.from,
+			to: payment.to,
+			amount: payment.amount,
+			hash: payment.hash,
+			timestamp: payment.timestamp,
 		});
 	}
 
@@ -156,8 +178,7 @@ export class PaymentListener extends DurableObject<Env> {
 				logger.info(`Invoice expired: ${activeInvoice.id}`, {
 					activeInvoice,
 				});
-				this.nanoWebsocket.unsubscribe(activeInvoice.payAddress);
-				this.removePendingInvoice(activeInvoice.id);
+				await this.removePendingInvoice(activeInvoice.id, activeInvoice.payAddress);
 			}
 		}
 		if (this.pendingInvoices.length > 0) {

src/durable/payment-listener.ts

@@ -0,0 +1,124 @@
+import { DurableObject } from 'cloudflare:workers';
+import { logger } from '../logger';
+
+export type PaymentNotification = {
+	from: string;
+	to: string;
+	amount: number;
+	hash: string;
+	timestamp: number;
+};
+
+/*
+ * PaymentNotifier implements a Durable Object that coordinates notifications for an individual invoice.
+ * Participants connect to the notifier using WebSockets, and the notifier broadcasts the payments for them.
+ * The notifier only start to respond websockets connections after started.
+ * It auto hibernates when no messages are being sent and resume when notify is called.
+ */
+export class PaymentNotifier extends DurableObject<Env> {
+	sessions = new Set<WebSocket>();
+	state: DurableObjectState;
+	storage: DurableObjectStorage;
+
+	constructor(ctx: DurableObjectState, env: Env) {
+		super(ctx, env);
+
+		this.state = ctx;
+		this.storage = ctx.storage;
+
+		this.state.getWebSockets().forEach((webSocket) => {
+			this.sessions.add(webSocket);
+		});
+	}
+
+	async fetch(request: Request) {
+		// We only accept WebSocket requests.
+		if (request.headers.get('Upgrade') != 'websocket') {
+			return new Response('expected websocket', { status: 400 });
+		}
+
+		const started = await this.storage.get('started');
+		if (started !== 'true') {
+			return new Response('not started', { status: 503 });
+		}
+
+		// To accept the WebSocket request, we create a WebSocketPair
+		const pair = new WebSocketPair();
+
+		// We're going to take pair[1] as our end, and return pair[0] to the client.
+		await this.handleSession(pair[1]);
+
+		// Now we return the other end of the pair to the client.
+		return new Response(null, { status: 101, webSocket: pair[0] });
+	}
+
+	async start() {
+		await this.storage.put('started', 'true');
+		logger.debug('Started payment notifier');
+	}
+
+	async stop() {
+		this.sessions.forEach((session) => {
+			session.close();
+		});
+		await this.clear();
+		logger.debug('Stopped payment notifier');
+	}
+
+	private async clear() {
+		this.sessions.clear();
+		await this.storage.deleteAll();
+	}
+
+	private async handleSession(webSocket: WebSocket) {
+		// Accept our end of the WebSocket. This tells the runtime that we'll be terminating the
+		// WebSocket in JavaScript, not sending it elsewhere.
+		this.state.acceptWebSocket(webSocket);
+
+		this.sessions.add(webSocket);
+
+		// Load the last 10 payments from the history stored on disk, and send them to the
+		// client.
+		const payments = await this.storage.list<Record<any, any>>({ reverse: true, limit: 10, prefix: 'payment_' });
+		[...payments.values()].forEach((value) => {
+			webSocket.send(JSON.stringify(value));
+		});
+	}
+
+	// On "close" and "error" events, remove the WebSocket from the sessions list and broadcast
+	// a quit message.
+	private async closeOrErrorHandler(webSocket: WebSocket) {
+		this.sessions.delete(webSocket);
+	}
+
+	// Implement DurableObject's webSocketClose event.
+	async webSocketClose(webSocket: WebSocket, code: number, reason: string, wasClean: boolean) {
+		this.closeOrErrorHandler(webSocket);
+	}
+
+	// Implement DurableObject's webSocketError event.
+	async webSocketError(webSocket: WebSocket, error: unknown) {
+		this.closeOrErrorHandler(webSocket);
+	}
+
+	// Broadcasts a payment to all clients.
+	private async broadcast(payment: PaymentNotification) {
+		// Iterate over all the sessions sending them messages.
+		this.sessions.forEach((session) => {
+			try {
+				session.send(JSON.stringify(payment));
+			} catch (err) {
+				// Whoops, this connection is dead. Remove it from the map
+				this.sessions.delete(session);
+			}
+		});
+	}
+
+	public async notify(payment: PaymentNotification) {
+		// Store the payment in the history.
+		await this.storage.put(`payment_${payment.hash}`, payment);
+
+		// Broadcast the payment to all clients.
+		this.broadcast(payment);
+	}
+}

src/durable/payment-notifier.ts

@@ -1,13 +1,13 @@
 import { invoiceRouter } from './invoice/invoice-router';
 import { logger } from './logger';
 import { queue } from './queues';
-import { UnauthorizedException } from './responses';
 import { isFalsyLike } from './utils';
 import { Router } from './utils/router';
 /*
  * Export our Durable Object classes here.
  */
 export { PaymentListener } from './durable/payment-listener';
+export { PaymentNotifier } from './durable/payment-notifier';
 
 /*
  * Our API router.
@@ -23,14 +23,6 @@ router.route('/invoices', invoiceRouter);
  */
 export default {
 	async fetch(request: Request<unknown, IncomingRequestCfProperties<unknown>>, env: Env, ctx: ExecutionContext): Promise<Response> {
-		// Check authorization
-		const authorizationHeader = request.headers.get('Authorization');
-		const bearerToken = authorizationHeader?.split(' ')[1];
-		const authorized = bearerToken === env.AUTH_TOKEN;
-		if (!authorized) {
-			return UnauthorizedException();
-		}
-
 		const localMode = !isFalsyLike(env.IS_LOCAL_MODE);
 		logger.setLocalDev(localMode);