rspace-online/modules/rflows/mod.ts

/**
 * Flows module — budget flows, river visualization, and treasury management.
 *
 * Proxies flow-service API calls and serves the FlowRiver visualization.
 */

import { Hono } from "hono";
import * as Automerge from "@automerge/automerge";
import { renderShell } from "../../server/shell";
import type { RSpaceModule } from "../../shared/module";
import { getModuleInfoList } from "../../shared/module";
import { verifyEncryptIDToken, extractToken } from "@encryptid/sdk/server";
import { renderLanding } from "./landing";
import type { SyncServer } from '../../server/local-first/sync-server';
import { flowsSchema, flowsDocId, type FlowsDoc, type SpaceFlow, type CanvasFlow } from './schemas';
import { demoNodes } from './lib/presets';
import { CoinbaseOnrampProvider } from './lib/coinbase-onramp';
import { OpenfortProvider } from './lib/openfort';

let _syncServer: SyncServer | null = null;
let _coinbaseOnramp: CoinbaseOnrampProvider | null = null;
let _openfort: OpenfortProvider | null = null;

const FLOW_SERVICE_URL = process.env.FLOW_SERVICE_URL || "http://payment-flow:3010";

function ensureDoc(space: string): FlowsDoc {
	const docId = flowsDocId(space);
	let doc = _syncServer!.getDoc<FlowsDoc>(docId);
	if (!doc) {
		doc = Automerge.change(Automerge.init<FlowsDoc>(), 'init', (d) => {
			const init = flowsSchema.init();
			d.meta = init.meta;
			d.meta.spaceSlug = space;
			d.spaceFlows = {};
			d.canvasFlows = {} as any;
			d.activeFlowId = '';
		});
		_syncServer!.setDoc(docId, doc);
	}
	// Migrate v1 → v2: add canvasFlows and activeFlowId
	if (!doc.canvasFlows || doc.meta.version < 2) {
		_syncServer!.changeDoc<FlowsDoc>(docId, 'migrate to v2', (d) => {
			if (!d.canvasFlows) d.canvasFlows = {} as any;
			if (!d.activeFlowId) d.activeFlowId = '' as any;
			d.meta.version = 2;
		});
		doc = _syncServer!.getDoc<FlowsDoc>(docId)!;
	}
	return doc;
}

const routes = new Hono();

// ─── Flow Service API proxy ─────────────────────────────
// These proxy to the payment-flow backend so the frontend
// can call them from the same origin.

routes.get("/api/flows", async (c) => {
	const owner = c.req.header("X-Owner-Address") || "";
	const space = c.req.query("space") || "";

	// If space filter provided, get flow IDs from Automerge doc
	if (space) {
		const doc = ensureDoc(space);
		const flowIds = Object.values(doc.spaceFlows).map((sf) => sf.flowId);
		if (flowIds.length === 0) return c.json([]);

		const flows = await Promise.all(
			flowIds.map(async (fid) => {
				try {
					const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${fid}`);
					if (res.ok) return await res.json();
				} catch {}
				return null;
			}),
		);
		return c.json(flows.filter(Boolean));
	}

	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows?owner=${encodeURIComponent(owner)}`);
	return c.json(await res.json(), res.status as any);
});

routes.get("/api/flows/:flowId", async (c) => {
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}`);
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows", async (c) => {
	// Auth-gated: require EncryptID token
	const token = extractToken(c.req.raw.headers);
	if (!token) return c.json({ error: "Authentication required" }, 401);

	let claims;
	try { claims = await verifyEncryptIDToken(token); } catch { return c.json({ error: "Invalid token" }, 401); }

	const body = await c.req.text();
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows`, {
		method: "POST",
		headers: {
			"Content-Type": "application/json",
			"X-Owner-Address": claims.sub,
		},
		body,
	});
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/deposit", async (c) => {
	const body = await c.req.text();
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/deposit`, {
		method: "POST",
		headers: { "Content-Type": "application/json" },
		body,
	});
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/withdraw", async (c) => {
	const body = await c.req.text();
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/withdraw`, {
		method: "POST",
		headers: { "Content-Type": "application/json" },
		body,
	});
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/activate", async (c) => {
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/activate`, { method: "POST" });
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/pause", async (c) => {
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/pause`, { method: "POST" });
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/funnels", async (c) => {
	const body = await c.req.text();
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/funnels`, {
		method: "POST",
		headers: { "Content-Type": "application/json" },
		body,
	});
	return c.json(await res.json(), res.status as any);
});

routes.post("/api/flows/:flowId/outcomes", async (c) => {
	const body = await c.req.text();
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/outcomes`, {
		method: "POST",
		headers: { "Content-Type": "application/json" },
		body,
	});
	return c.json(await res.json(), res.status as any);
});

routes.get("/api/flows/:flowId/transactions", async (c) => {
	const res = await fetch(`${FLOW_SERVICE_URL}/api/flows/${c.req.param("flowId")}/transactions`);
	return c.json(await res.json(), res.status as any);
});

// ─── User on-ramp (email → wallet → widget) ─────────────

routes.post("/api/flows/user-onramp", async (c) => {
	try {
		const { email, fiatAmount, fiatCurrency, returnUrl, provider: reqProvider } = await c.req.json();
		if (!email || !fiatAmount || !fiatCurrency) {
			return c.json({ error: "email, fiatAmount, and fiatCurrency are required" }, 400);
		}

		if (!_openfort) return c.json({ error: "Openfort not configured" }, 503);

		// Determine provider: explicit request > env default > auto-detect
		const provider = reqProvider
			|| process.env.ONRAMP_PROVIDER
			|| (_coinbaseOnramp ? 'coinbase' : 'transak');

		// 1. Find or create Openfort smart wallet for this user (one wallet per email)
		const wallet = await _openfort.findOrCreateWallet(`user:${email}`, {
			type: 'user-onramp',
			email,
		});

		const sessionId = crypto.randomUUID();
		let widgetUrl: string;

		if (provider === 'coinbase') {
			// 2a. Coinbase: server-side session → widget URL
			if (!_coinbaseOnramp) return c.json({ error: "Coinbase Onramp not configured" }, 503);
			const session = await _coinbaseOnramp.createSession({
				walletAddress: wallet.address,
				fiatAmount,
				fiatCurrency,
				partnerUserRef: `user-${sessionId}`,
				redirectUrl: returnUrl,
			});
			widgetUrl = session.onrampUrl;
		} else {
			// 2b. Transak: build widget URL server-side
			const transakApiKey = process.env.TRANSAK_API_KEY;
			if (!transakApiKey) return c.json({ error: "Transak not configured" }, 503);
			const transakEnv = process.env.TRANSAK_ENV || 'PRODUCTION';
			const baseUrl = transakEnv === 'PRODUCTION'
				? 'https://global.transak.com'
				: 'https://global-stg.transak.com';
			const params = new URLSearchParams({
				apiKey: transakApiKey,
				environment: transakEnv,
				cryptoCurrencyCode: 'USDC',
				network: 'base',
				defaultCryptoCurrency: 'USDC',
				walletAddress: wallet.address,
				partnerOrderId: `user-${sessionId}`,
				email,
				themeColor: '6366f1',
				hideMenu: 'true',
			});
			if (returnUrl) params.set('redirectURL', returnUrl);
			widgetUrl = `${baseUrl}?${params}`;
		}

		console.log(`[rflows] On-ramp session created: provider=${provider} session=${sessionId} wallet=${wallet.address}`);

		// Non-fatal side-effect: create fund claim → sends email via EncryptID
		const encryptidServiceKey = process.env.ENCRYPTID_SERVICE_KEY;
		if (encryptidServiceKey) {
			const encryptidUrl = process.env.ENCRYPTID_URL || 'https://auth.rspace.online';
			try {
				await fetch(`${encryptidUrl}/api/internal/fund-claims`, {
					method: 'POST',
					headers: {
						'Content-Type': 'application/json',
						'X-Service-Key': encryptidServiceKey,
					},
					body: JSON.stringify({
						email,
						walletAddress: wallet.address,
						openfortPlayerId: wallet.playerId,
						fiatAmount: String(fiatAmount),
						fiatCurrency,
						sessionId,
						provider,
					}),
				});
				console.log(`[rflows] Fund claim created for ${email}`);
			} catch (err) {
				console.error('[rflows] Failed to create fund claim:', err);
			}
		}

		return c.json({
			success: true,
			sessionId,
			widgetUrl,
			walletAddress: wallet.address,
			provider,
			isNewUser: true,
		});
	} catch (err) {
		console.error("[rflows] user-onramp failed:", err);
		const message = err instanceof Error ? err.message : "Unknown error";
		return c.json({ error: message }, 500);
	}
});

// ─── On-ramp config ──────────────────────────────────────

routes.get("/api/onramp/config", (c) => {
	const available: string[] = [];
	if (process.env.COINBASE_CDP_PROJECT_ID) available.push("coinbase");
	if (process.env.TRANSAK_API_KEY || !process.env.COINBASE_CDP_PROJECT_ID) available.push("transak");
	return c.json({
		provider: process.env.ONRAMP_PROVIDER || (process.env.COINBASE_CDP_PROJECT_ID ? "coinbase" : "transak"),
		available,
		// Transak fields (only needed if provider=transak)
		apiKey: process.env.TRANSAK_API_KEY || "",
		environment: process.env.TRANSAK_ENV || "PRODUCTION",
	});
});

// Legacy endpoint — keep for backwards compat
routes.get("/api/transak/config", (c) => {
	return c.json({
		apiKey: process.env.TRANSAK_API_KEY || "",
		environment: process.env.TRANSAK_ENV || "PRODUCTION",
	});
});

routes.post("/api/transak/webhook", async (c) => {
	let body: any;
	try { body = await c.req.json(); } catch { return c.json({ error: "Invalid JSON" }, 400); }

	// HMAC verification — if TRANSAK_WEBHOOK_SECRET is set, validate signature
	const webhookSecret = process.env.TRANSAK_WEBHOOK_SECRET;
	if (webhookSecret) {
		const signature = c.req.header("x-transak-signature") || "";
		const { createHmac } = await import("crypto");
		// Re-serialize for HMAC (Transak signs the raw JSON body)
		const expected = createHmac("sha256", webhookSecret).update(JSON.stringify(body)).digest("hex");
		if (signature !== expected) {
			console.error("[Transak] Invalid webhook signature");
			return c.json({ error: "Invalid signature" }, 401);
		}
	}

	const { webhookData } = body;

	// Ack non-completion events (Transak sends multiple status updates)
	if (!webhookData || webhookData.status !== "COMPLETED") {
		return c.json({ ok: true });
	}

	const { partnerOrderId, cryptoAmount, cryptocurrency, network } = webhookData;
	if (!partnerOrderId || cryptocurrency !== "USDC" || !network?.toLowerCase().includes("base")) {
		return c.json({ error: "Invalid webhook data" }, 400);
	}

	// partnerOrderId format: "flowId:funnelId" or "flowId" (uses env default)
	const [flowId, funnelId] = partnerOrderId.split(":");
	if (!flowId) return c.json({ error: "Missing flowId in partnerOrderId" }, 400);
	const resolvedFunnelId = funnelId || process.env.FUNNEL_ID || "";
	if (!resolvedFunnelId) return c.json({ error: "Missing funnelId" }, 400);

	// Convert crypto amount to USDC units (6 decimals)
	const amountUnits = Math.round(parseFloat(cryptoAmount) * 1e6).toString();

	const depositUrl = `${FLOW_SERVICE_URL}/api/flows/${flowId}/deposit`;
	const res = await fetch(depositUrl, {
		method: "POST",
		headers: { "Content-Type": "application/json" },
		body: JSON.stringify({
			amount: amountUnits,
			source: "card",
			funnelId: resolvedFunnelId,
		}),
	});

	if (!res.ok) {
		console.error(`[Transak] Deposit failed: ${await res.text()}`);
		return c.json({ error: "Deposit failed" }, 500);
	}

	console.log(`[Transak] Deposit OK: flow=${flowId} amount=${amountUnits} USDC`);
	return c.json({ ok: true });
});

// ─── Space-flow association endpoints ────────────────────

routes.post("/api/space-flows", async (c) => {
	const token = extractToken(c.req.raw.headers);
	if (!token) return c.json({ error: "Authentication required" }, 401);

	let claims;
	try { claims = await verifyEncryptIDToken(token); } catch { return c.json({ error: "Invalid token" }, 401); }

	const { space, flowId } = await c.req.json();
	if (!space || !flowId) return c.json({ error: "space and flowId required" }, 400);

	const docId = flowsDocId(space);
	ensureDoc(space);
	_syncServer!.changeDoc<FlowsDoc>(docId, 'add space flow', (d) => {
		const key = `${space}:${flowId}`;
		if (!d.spaceFlows[key]) {
			d.spaceFlows[key] = { id: key, spaceSlug: space, flowId, addedBy: claims.sub, createdAt: Date.now() };
		}
	});
	return c.json({ ok: true });
});

routes.delete("/api/space-flows/:flowId", async (c) => {
	const token = extractToken(c.req.raw.headers);
	if (!token) return c.json({ error: "Authentication required" }, 401);

	try { await verifyEncryptIDToken(token); } catch { return c.json({ error: "Invalid token" }, 401); }

	const flowId = c.req.param("flowId");
	const space = c.req.query("space") || "";
	if (!space) return c.json({ error: "space query param required" }, 400);

	const docId = flowsDocId(space);
	const doc = _syncServer!.getDoc<FlowsDoc>(docId);
	if (doc) {
		const key = `${space}:${flowId}`;
		if (doc.spaceFlows[key]) {
			_syncServer!.changeDoc<FlowsDoc>(docId, 'remove space flow', (d) => {
				delete d.spaceFlows[key];
			});
		}
	}
	return c.json({ ok: true });
});

// ─── Page routes ────────────────────────────────────────

const flowsScripts = `
	<script src="https://cdn.jsdelivr.net/npm/lz-string@1.5.0/libs/lz-string.min.js"></script>
	<script type="module" src="/modules/rflows/folk-flows-app.js"></script>
	<script type="module" src="/modules/rflows/folk-flow-river.js"></script>`;

const flowsStyles = `<link rel="stylesheet" href="/modules/rflows/flows.css">`;

// Landing page (also serves demo via centralized /demo → space="demo" rewrite)
routes.get("/", (c) => {
	const spaceSlug = c.req.param("space") || "demo";
	return c.html(renderShell({
		title: `${spaceSlug} — Flows | rSpace`,
		moduleId: "rflows",
		spaceSlug,
		modules: getModuleInfoList(),
		theme: "dark",
		body: `<folk-flows-app space="${spaceSlug}"${spaceSlug === "demo" ? ' mode="demo"' : ''}></folk-flows-app>`,
		scripts: flowsScripts,
		styles: flowsStyles,
	}));
});

// Flow detail — specific flow from API
routes.get("/flow/:flowId", (c) => {
	const spaceSlug = c.req.param("space") || "demo";
	const flowId = c.req.param("flowId");
	return c.html(renderShell({
		title: `Flow — rFlows | rSpace`,
		moduleId: "rflows",
		spaceSlug,
		modules: getModuleInfoList(),
		theme: "dark",
		styles: flowsStyles,
		body: `<folk-flows-app space="${spaceSlug}" flow-id="${flowId}"></folk-flows-app>`,
		scripts: flowsScripts,
	}));
});

// ── Seed template data ──

function seedTemplateFlows(space: string) {
	if (!_syncServer) return;
	const doc = ensureDoc(space);

	// Seed SpaceFlow association if empty
	if (Object.keys(doc.spaceFlows).length === 0) {
		const docId = flowsDocId(space);
		const now = Date.now();
		const flowId = crypto.randomUUID();

		_syncServer.changeDoc<FlowsDoc>(docId, 'seed template flow', (d) => {
			d.spaceFlows[flowId] = {
				id: flowId, spaceSlug: space, flowId: 'demo',
				addedBy: 'did:demo:seed', createdAt: now,
			};
		});
	}

	// Seed a canvas flow with demoNodes if none exist
	if (Object.keys(doc.canvasFlows || {}).length === 0) {
		const docId = flowsDocId(space);
		const now = Date.now();
		const canvasFlowId = crypto.randomUUID();

		const seedFlow: CanvasFlow = {
			id: canvasFlowId,
			name: 'TBFF Demo Flow',
			nodes: demoNodes.map((n) => ({ ...n, data: { ...n.data } })),
			createdAt: now,
			updatedAt: now,
			createdBy: 'did:demo:seed',
		};

		_syncServer.changeDoc<FlowsDoc>(docId, 'seed canvas flow', (d) => {
			d.canvasFlows[canvasFlowId] = seedFlow as any;
			d.activeFlowId = canvasFlowId as any;
		});

		console.log(`[Flows] Template seeded for "${space}": 1 canvas flow + association`);
	}
}

export const flowsModule: RSpaceModule = {
	id: "rflows",
	name: "rFlows",
	icon: "🌊",
	description: "Budget flows, river visualization, and treasury management",
	publicWrite: true,
	scoping: { defaultScope: 'space', userConfigurable: false },
	docSchemas: [{ pattern: '{space}:flows:data', description: 'Space flow associations', init: flowsSchema.init }],
	routes,
	landingPage: renderLanding,
	seedTemplate: seedTemplateFlows,
	async onInit(ctx) {
		_syncServer = ctx.syncServer;

		// Initialize on-ramp providers if env vars are set
		if (process.env.COINBASE_CDP_KEY_ID && process.env.COINBASE_CDP_KEY_SECRET && process.env.COINBASE_CDP_PROJECT_ID) {
			_coinbaseOnramp = new CoinbaseOnrampProvider({
				apiKeyId: process.env.COINBASE_CDP_KEY_ID,
				apiKeySecret: process.env.COINBASE_CDP_KEY_SECRET,
				projectId: process.env.COINBASE_CDP_PROJECT_ID,
				environment: (process.env.COINBASE_ENVIRONMENT as 'sandbox' | 'production') || 'production',
			});
			console.log('[rflows] Coinbase Onramp provider initialized');
		}

		if (process.env.OPENFORT_API_KEY && process.env.OPENFORT_PUBLISHABLE_KEY) {
			_openfort = new OpenfortProvider({
				apiKey: process.env.OPENFORT_API_KEY,
				publishableKey: process.env.OPENFORT_PUBLISHABLE_KEY,
				chainId: parseInt(process.env.BASE_CHAIN_ID || '8453', 10),
			});
			console.log('[rflows] Openfort provider initialized');
		}
	},
	standaloneDomain: "rflows.online",
	feeds: [
		{
			id: "treasury-flows",
			name: "Treasury Flows",
			kind: "economic",
			description: "Budget flow states, deposits, withdrawals, and funnel allocations",
			filterable: true,
		},
		{
			id: "transactions",
			name: "Transaction Stream",
			kind: "economic",
			description: "Real-time deposit and withdrawal events",
		},
	],
	acceptsFeeds: ["governance", "data"],
	outputPaths: [
		{ path: "budgets", name: "Budgets", icon: "💰", description: "Budget allocations and funnels" },
		{ path: "flows", name: "Flows", icon: "🌊", description: "Revenue and resource flow visualizations" },
	],
	subPageInfos: [
		{
			path: "flow",
			title: "Flow Viewer",
			icon: "🌊",
			tagline: "rFlows Tool",
			description: "Visualize a single budget flow — deposits, withdrawals, funnel allocations, and real-time balance. Drill into transactions and manage outcomes.",
			features: [
				{ icon: "📈", title: "River Visualization", text: "See funds flow through funnels and outcomes as an animated river diagram." },
				{ icon: "💸", title: "Deposits & Withdrawals", text: "Track every transaction with full history and on-chain verification." },
				{ icon: "🎯", title: "Outcome Tracking", text: "Define funding outcomes and monitor how capital reaches its destination." },
			],
		},
	],
};