/** * 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(docId); if (!doc) { doc = Automerge.change(Automerge.init(), '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(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(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 || 'STAGING_KEY'; const transakEnv = process.env.TRANSAK_ENV || 'STAGING'; 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 || "STAGING_KEY", environment: process.env.TRANSAK_ENV || "STAGING", }); }); // Legacy endpoint — keep for backwards compat routes.get("/api/transak/config", (c) => { return c.json({ apiKey: process.env.TRANSAK_API_KEY || "STAGING_KEY", environment: process.env.TRANSAK_ENV || "STAGING", }); }); 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(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(docId); if (doc) { const key = `${space}:${flowId}`; if (doc.spaceFlows[key]) { _syncServer!.changeDoc(docId, 'remove space flow', (d) => { delete d.spaceFlows[key]; }); } } return c.json({ ok: true }); }); // ─── Page routes ──────────────────────────────────────── const flowsScripts = ` `; const flowsStyles = ``; // 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: ``, 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: ``, 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(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(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') || 'sandbox', }); 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 || '84532', 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." }, ], }, ], };