diff --git a/packages/inference-server/src/lib/url-prefetch.ts b/packages/inference-server/src/lib/url-prefetch.ts index 1fdb3cf..d8055ae 100644 --- a/packages/inference-server/src/lib/url-prefetch.ts +++ b/packages/inference-server/src/lib/url-prefetch.ts @@ -32,6 +32,18 @@ const FETCH_TIMEOUT_MS = 5_000; const MAX_BYTES = 50 * 1024; const MAX_EXTRACT_CHARS = 30 * 1024; const UA = "WaveX-OS-Onboarding/1.0 (+https://github.com/aimerdoux/wavex-os)"; +// Minimum readable text (after stripping the [META] prefix) for a page to +// count as "real" content. Below this, the site is almost certainly parked, +// a redirect stub, or JS-rendered with no SSR — feeding it to the model just +// invites a confident hallucination, so we mark it FETCH FAILED instead. +const MIN_CONTENT_CHARS = 120; +// `` — common for parked +// domains and some real sites. fetch() follows HTTP redirects but not this. +const META_REFRESH_REGEX = /]+http-equiv=["']?refresh["']?[^>]*content=["'][^"']*url=([^"'>\s]+)/i; +// Parked / placeholder fingerprints (IONOS, GoDaddy, Sedo, generic "buy this +// domain" pages). If the only content we can read is a parking page, the model +// must not infer a real company from it. +const PARKED_RE = /\b(este dominio ya|buy this domain|this domain (is|may be) for sale|domain is parked|parkingcrew|sedoparking|godaddy\.com\/domains|hugedomains|is registered with ionos|defaultsite|under construction|coming soon)\b/i; const PRIVATE_HOST_PATTERNS = [ /^localhost$/i, @@ -94,7 +106,10 @@ interface FetchResult { reason?: string; } -async function fetchOne(url: URL): Promise { +/** Fetch raw HTML (byte-capped) for a single URL. Returns the decoded string + * or a failure reason. No extraction — callers run htmlToText / meta-refresh + * detection on the raw markup. */ +async function fetchRawHtml(url: URL): Promise<{ ok: true; raw: string } | { ok: false; reason: string }> { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS); try { @@ -104,18 +119,13 @@ async function fetchOne(url: URL): Promise { signal: controller.signal, redirect: "follow", }); - if (!resp.ok) { - return { url: url.toString(), status: "failed", reason: `HTTP ${resp.status}` }; - } + if (!resp.ok) return { ok: false, reason: `HTTP ${resp.status}` }; const ct = resp.headers.get("content-type") ?? ""; - if (!/text\/html|application\/xhtml/i.test(ct)) { - return { url: url.toString(), status: "failed", reason: `non-html content-type: ${ct}` }; - } - // Cap bytes by reading until limit + if (!/text\/html|application\/xhtml/i.test(ct)) return { ok: false, reason: `non-html content-type: ${ct}` }; const reader = resp.body?.getReader(); if (!reader) { const txt = await resp.text(); - return { url: url.toString(), status: "ok", body: htmlToText(txt.slice(0, MAX_BYTES)) }; + return { ok: true, raw: txt.slice(0, MAX_BYTES) }; } const chunks: Uint8Array[] = []; let received = 0; @@ -130,16 +140,46 @@ async function fetchOne(url: URL): Promise { const merged = new Uint8Array(concatLen); let off = 0; for (const c of chunks) { merged.set(c, off); off += c.length; } - const raw = new TextDecoder("utf-8", { fatal: false }).decode(merged.slice(0, MAX_BYTES)); - return { url: url.toString(), status: "ok", body: htmlToText(raw) }; + return { ok: true, raw: new TextDecoder("utf-8", { fatal: false }).decode(merged.slice(0, MAX_BYTES)) }; } catch (e) { const reason = (e as Error).name === "AbortError" ? `timeout after ${FETCH_TIMEOUT_MS}ms` : (e as Error).message; - return { url: url.toString(), status: "failed", reason }; + return { ok: false, reason }; } finally { clearTimeout(timer); } } +async function fetchOne(url: URL): Promise { + const first = await fetchRawHtml(url); + if (!first.ok) return { url: url.toString(), status: "failed", reason: first.reason }; + let raw = first.raw; + + // Follow ONE meta-refresh hop (parked domains + some real sites use it; plain + // fetch() doesn't). Resolve the target relative to the current URL, validate + // it's a safe public host, and re-fetch. + const mr = raw.match(META_REFRESH_REGEX); + if (mr) { + try { + const target = new URL(mr[1].replace(/&/g, "&"), url); + if (isSafePublicUrl(target.toString())) { + const hop = await fetchRawHtml(target); + if (hop.ok) raw = hop.raw; + } + } catch { /* keep the first page */ } + } + + const body = htmlToText(raw); + // Strip the [META] ... prefix when measuring real body text. + const textOnly = body.replace(/^\[META\][^\n]*\n+/, "").trim(); + if (textOnly.length < MIN_CONTENT_CHARS) { + return { url: url.toString(), status: "failed", reason: "no readable content (site may be parked, empty, or JavaScript-rendered)" }; + } + if (PARKED_RE.test(body)) { + return { url: url.toString(), status: "failed", reason: "domain appears parked / not a live site" }; + } + return { url: url.toString(), status: "ok", body }; +} + /** Pre-fetch every public http(s) URL in `prompt` and return a new prompt * with FETCHED / FETCH FAILED markers injected before each URL's first * appearance. Idempotent — if the prompt already contains [FETCHED] markers diff --git a/packages/wavex-os-server/src/index.ts b/packages/wavex-os-server/src/index.ts index f12ca6a..c3e62bd 100644 --- a/packages/wavex-os-server/src/index.ts +++ b/packages/wavex-os-server/src/index.ts @@ -58,7 +58,10 @@ import { registerGitHubReposRoute } from "./routes/github-repos.js"; import { registerMissionControlRoutes } from "./routes/mission-control.js"; import { startReferralEmailBScheduler } from "./jobs/referral-email-b.js"; import { startProfessionalReengagementScheduler } from "./jobs/professional-reengagement.js"; +import { startBookingIntentCleanupScheduler } from "./jobs/booking-intent-cleanup.js"; +import { runAbandonedBookingRecoveryJob } from "./jobs/abandoned-booking-recovery.js"; import { registerReengagementRoutes } from "./routes/reengagement.js"; +import { registerBookingRecoveryRoute } from "./routes/booking-recovery.js"; let bootstrapped = false; function bootstrap(): void { @@ -119,8 +122,16 @@ export function registerWavexOsRoutes(app: FastifyInstance): void { registerGitHubReposRoute(app); registerMissionControlRoutes(app); registerReengagementRoutes(app); + registerBookingRecoveryRoute(app); startReferralEmailBScheduler(); startProfessionalReengagementScheduler(); + startBookingIntentCleanupScheduler(); + // One-shot: diagnose + optionally send abandoned booking recovery nudges. + // Safe by default (WAVEX_ABANDONED_BOOKING_DRY_RUN=true). Set to "false" + // after CEO approval to actually send. idempotent via nudge log unique constraint. + void runAbandonedBookingRecoveryJob().catch((err) => + console.error("[abandoned-booking-recovery] startup run failed:", err), + ); } export { applyStateBridge, getInstanceDir, getOnboardingDir, getWavexDataRoot } from "./state-bridge.js"; diff --git a/packages/wavex-os-server/src/jobs/abandoned-booking-recovery.ts b/packages/wavex-os-server/src/jobs/abandoned-booking-recovery.ts new file mode 100644 index 0000000..192ab05 --- /dev/null +++ b/packages/wavex-os-server/src/jobs/abandoned-booking-recovery.ts @@ -0,0 +1,264 @@ +/** Abandoned booking recovery — one-time Telegram nudge (WAVAAAA-1197) + * + * Runs ONCE at server startup. Queries wavex_os_list_abandoned_booking_candidates() + * for cancelled booking_intents that have a resolvable Telegram chat_id, composes + * a recovery message, and either logs (dry-run) or sends via the Telegram Bot API. + * + * Context: users whose booking_intents were auto-cancelled before the Stripe + * checkout flow was added in bfeff1f5 had no payment path and never converted. + * This job gives them a single nudge informing them booking is now available. + * + * Dry-run gating (default ON — safe to deploy without CEO approval): + * WAVEX_ABANDONED_BOOKING_DRY_RUN — default "true". Set to "false" (case- + * insensitive) to send real Telegram messages. While true the job still + * selects candidates and logs the recoverable count so the CEO can review + * before enabling live sends. + * + * Recovery message: + * Default copy lives in DEFAULT_RECOVERY_MESSAGE below. Override via + * WAVEX_ABANDONED_BOOKING_MESSAGE (supports {{experience_name}} substitution). + * WAVEX_BOOKING_URL sets the booking link included in the message. + * + * Required env vars: + * SUPABASE_URL — PostgREST endpoint + * SUPABASE_SERVICE_ROLE_KEY — service-role JWT + * TELEGRAM_BOT_TOKEN — concierge bot token (for real sends) + * Optional env vars: + * WAVEX_ABANDONED_BOOKING_DRY_RUN — "true" | "false" (default "true") + * WAVEX_BOOKING_URL — booking URL included in message + * WAVEX_ABANDONED_BOOKING_MESSAGE — custom message template + */ + +interface SupabaseConfig { + url: string; + key: string; +} + +interface AbandonedCandidate { + intent_id: string; + telegram_user_id: string; + chat_id: string; + experience_name: string | null; + experience_price_cents: number; + currency: string; + created_at: string; + cancelled_at: string | null; +} + +const DEFAULT_RECOVERY_MESSAGE = + "Hey! Your booking request{{experience_part}} was cancelled due to a payment " + + "system issue on our end — we're really sorry about that.\n\n" + + "Great news: booking is back up and working! You can complete your booking here:\n" + + "{{booking_url}}\n\n" + + "Reply here if you have any questions and I'll sort it out for you! 🙌"; + +function supabaseConfig(): SupabaseConfig | null { + const url = process.env.SUPABASE_URL; + const key = process.env.SUPABASE_SERVICE_ROLE_KEY; + if (!url || !key) return null; + return { url, key }; +} + +function isDryRun(): boolean { + const raw = (process.env.WAVEX_ABANDONED_BOOKING_DRY_RUN ?? "true").toLowerCase(); + return raw !== "false"; +} + +function renderMessage(candidate: AbandonedCandidate): string { + const template = + process.env.WAVEX_ABANDONED_BOOKING_MESSAGE ?? DEFAULT_RECOVERY_MESSAGE; + const bookingUrl = process.env.WAVEX_BOOKING_URL ?? "[booking link coming soon]"; + const experiencePart = candidate.experience_name + ? ` for ${candidate.experience_name}` + : ""; + return template + .replace(/{{experience_part}}/g, experiencePart) + .replace(/{{experience_name}}/g, candidate.experience_name ?? "your experience") + .replace(/{{booking_url}}/g, bookingUrl); +} + +async function listCandidates(cfg: SupabaseConfig): Promise { + const res = await fetch( + `${cfg.url}/rest/v1/rpc/wavex_os_list_abandoned_booking_candidates`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + apikey: cfg.key, + Authorization: `Bearer ${cfg.key}`, + }, + body: "{}", + }, + ); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + throw new Error( + `wavex_os_list_abandoned_booking_candidates failed: ${res.status} ${detail}`, + ); + } + return (await res.json().catch(() => [])) as AbandonedCandidate[]; +} + +async function recordNudge( + cfg: SupabaseConfig, + args: { + intent_id: string; + telegram_user_id: string; + chat_id: string; + message_text: string; + message_id: string | null; + dry_run: boolean; + }, +): Promise { + const res = await fetch( + `${cfg.url}/rest/v1/rpc/wavex_os_record_abandoned_booking_nudge`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + apikey: cfg.key, + Authorization: `Bearer ${cfg.key}`, + }, + body: JSON.stringify({ + p_intent_id: args.intent_id, + p_telegram_user_id: args.telegram_user_id, + p_chat_id: args.chat_id, + p_message_text: args.message_text, + p_message_id: args.message_id, + p_dry_run: args.dry_run, + }), + }, + ); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + console.error( + `[abandoned-booking-recovery] failed to record nudge intent=${args.intent_id}: ${res.status} ${detail}`, + ); + } +} + +interface TelegramSendResult { + ok: boolean; + message_id?: string; + error?: string; +} + +async function sendTelegram( + token: string, + chatId: string, + text: string, +): Promise { + const ctrl = new AbortController(); + const t = setTimeout(() => ctrl.abort(), 10_000); + try { + const res = await fetch( + `https://api.telegram.org/bot${encodeURIComponent(token)}/sendMessage`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ chat_id: chatId, text }), + signal: ctrl.signal, + }, + ); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + return { ok: false, error: `${res.status} ${detail}` }; + } + const body = (await res.json().catch(() => null)) as + | { ok: boolean; result?: { message_id?: number } } + | null; + const mid = body?.result?.message_id; + return { ok: true, message_id: mid !== undefined ? String(mid) : undefined }; + } catch (e) { + return { ok: false, error: e instanceof Error ? e.message : String(e) }; + } finally { + clearTimeout(t); + } +} + +export interface AbandonedBookingRecoveryResult { + recoverable: number; // total candidates found (with reachable chat_id) + sent: number; + dryRun: number; + skipped: number; + errors: number; +} + +export async function runAbandonedBookingRecoveryJob(): Promise { + const cfg = supabaseConfig(); + if (!cfg) { + console.warn( + "[abandoned-booking-recovery] Supabase not configured — skipping run", + ); + return { recoverable: 0, sent: 0, dryRun: 0, skipped: 0, errors: 0 }; + } + + const dryRun = isDryRun(); + const token = process.env.TELEGRAM_BOT_TOKEN; + + let candidates: AbandonedCandidate[]; + try { + candidates = await listCandidates(cfg); + } catch (err) { + console.error("[abandoned-booking-recovery] candidate query failed:", err); + return { recoverable: 0, sent: 0, dryRun: 0, skipped: 0, errors: 1 }; + } + + console.info( + `[abandoned-booking-recovery] recoverable=${candidates.length} dryRun=${dryRun}`, + ); + + let sent = 0; + let dryRunRecorded = 0; + let skipped = 0; + let errors = 0; + + for (const c of candidates) { + const text = renderMessage(c); + + let send: TelegramSendResult = { ok: true }; + if (dryRun) { + console.log( + `[abandoned-booking-recovery] [dry-run] intent=${c.intent_id} chat=${c.chat_id} msg=${JSON.stringify(text)}`, + ); + dryRunRecorded++; + } else if (!token) { + console.warn( + "[abandoned-booking-recovery] TELEGRAM_BOT_TOKEN not set — cannot send; skipping", + ); + skipped++; + continue; + } else { + send = await sendTelegram(token, c.chat_id, text); + if (!send.ok) { + console.error( + `[abandoned-booking-recovery] send failed intent=${c.intent_id}: ${send.error}`, + ); + errors++; + continue; + } + sent++; + } + + await recordNudge(cfg, { + intent_id: c.intent_id, + telegram_user_id: c.telegram_user_id, + chat_id: c.chat_id, + message_text: text, + message_id: send.message_id ?? null, + dry_run: dryRun, + }); + } + + const result: AbandonedBookingRecoveryResult = { + recoverable: candidates.length, + sent, + dryRun: dryRunRecorded, + skipped, + errors, + }; + console.info( + `[abandoned-booking-recovery] run complete: ${JSON.stringify(result)}`, + ); + return result; +} diff --git a/packages/wavex-os-server/src/jobs/booking-intent-cleanup.ts b/packages/wavex-os-server/src/jobs/booking-intent-cleanup.ts new file mode 100644 index 0000000..a2a6f5a --- /dev/null +++ b/packages/wavex-os-server/src/jobs/booking-intent-cleanup.ts @@ -0,0 +1,84 @@ +/** Booking-intent stale cleanup (WAVAAAA-1195) + * + * Runs every 5 minutes. Calls wavex_os_cancel_stale_booking_intents() to + * cancel booking_intents that have been in status='pending' for longer than + * 30 minutes. + * + * Critically: intents in status='pending_payment' (user is in Stripe redirect) + * are NOT cancelled. Only 'pending' intents (where no checkout was ever + * initiated) time out. This prevents cancelling live payment sessions. + * + * Required env vars: + * SUPABASE_URL — PostgREST endpoint + * SUPABASE_SERVICE_ROLE_KEY — service-role JWT + * Optional env vars: + * WAVEX_BOOKING_INTENT_TIMEOUT_MINUTES — integer (default 30) + */ + +interface SupabaseConfig { + url: string; + key: string; +} + +function supabaseConfig(): SupabaseConfig | null { + const url = process.env.SUPABASE_URL; + const key = process.env.SUPABASE_SERVICE_ROLE_KEY; + if (!url || !key) return null; + return { url, key }; +} + +function envInt(name: string, fallback: number): number { + const raw = process.env[name]; + if (!raw) return fallback; + const n = Number.parseInt(raw, 10); + return Number.isFinite(n) && n > 0 ? n : fallback; +} + +async function cancelStaleIntents(cfg: SupabaseConfig): Promise { + const timeoutMinutes = envInt("WAVEX_BOOKING_INTENT_TIMEOUT_MINUTES", 30); + const res = await fetch( + `${cfg.url}/rest/v1/rpc/wavex_os_cancel_stale_booking_intents`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + apikey: cfg.key, + Authorization: `Bearer ${cfg.key}`, + }, + body: JSON.stringify({ p_older_than_minutes: timeoutMinutes }), + }, + ); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + throw new Error(`wavex_os_cancel_stale_booking_intents failed: ${res.status} ${detail}`); + } + const data = (await res.json().catch(() => null)) as { cancelled?: number } | null; + return data?.cancelled ?? 0; +} + +let cleanupTimer: ReturnType | null = null; + +export function startBookingIntentCleanupScheduler(): void { + if (cleanupTimer) return; + + const INTERVAL_MS = 5 * 60 * 1000; // every 5 minutes + + const run = async () => { + const cfg = supabaseConfig(); + if (!cfg) return; + try { + const cancelled = await cancelStaleIntents(cfg); + if (cancelled > 0) { + console.info(`[booking-intent-cleanup] cancelled ${cancelled} stale pending intent(s)`); + } + } catch (e) { + console.error( + `[booking-intent-cleanup] run threw: ${e instanceof Error ? e.message : String(e)}`, + ); + } + }; + + // Run once immediately, then on the interval. + void run(); + cleanupTimer = setInterval(() => void run(), INTERVAL_MS); +} diff --git a/packages/wavex-os-server/src/routes/auth-events.ts b/packages/wavex-os-server/src/routes/auth-events.ts index c7c32cb..ac922ea 100644 --- a/packages/wavex-os-server/src/routes/auth-events.ts +++ b/packages/wavex-os-server/src/routes/auth-events.ts @@ -26,6 +26,8 @@ const bodySchema = z.object({ type AuthEventBody = z.infer; // ─── Supabase write ──────────────────────────────────────────────────────── +// Uses wavex_os_record_auth_event RPC (public schema SECURITY DEFINER) because +// the wavex_os schema is intentionally not exposed via PostgREST. async function writeAuthEvent(row: { user_id: string; @@ -42,25 +44,30 @@ async function writeAuthEvent(row: { console.warn("[auth-events] Supabase not configured — event not persisted"); return null; } - const res = await fetch(`${url}/rest/v1/wavex_os.auth_events`, { + const res = await fetch(`${url}/rest/v1/rpc/wavex_os_record_auth_event`, { method: "POST", headers: { "Content-Type": "application/json", - "Prefer": "return=representation,resolution=ignore-duplicates", apikey: key, Authorization: `Bearer ${key}`, }, - body: JSON.stringify(row), + body: JSON.stringify({ + p_user_id: row.user_id, + p_email: row.email ?? null, + p_event_type: row.event_type, + p_utm_campaign: row.utm_campaign ?? null, + p_utm_source: row.utm_source ?? null, + p_ref: row.ref ?? null, + p_resend_fired: row.resend_fired, + }), }); if (!res.ok) { const detail = await res.text().catch(() => ""); - // 409 / duplicate constraint → idempotent, not an error. - if (res.status === 409) return null; console.error(`[auth-events] DB write failed: ${res.status} ${detail}`); return null; } - const records = (await res.json().catch(() => [])) as Array<{ id: string }>; - return records[0] ?? null; + const id = (await res.json().catch(() => null)) as string | null; + return id ? { id } : null; } // ─── Resend attribution ──────────────────────────────────────────────────── @@ -155,19 +162,19 @@ export function registerAuthEventsRoute(app: FastifyInstance): void { return reply.status(503).send({ ok: false, error: "Supabase not configured" }); } - const res = await fetch( - `${url}/rest/v1/wavex_os.auth_events` + - `?utm_campaign=eq.${encodeURIComponent(campaign)}` + - `&event_type=eq.signup_confirmed` + - `&created_at=gte.${encodeURIComponent(new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString())}` + - `&select=user_id`, - { headers: { apikey: key, Authorization: `Bearer ${key}` } }, - ); + const res = await fetch(`${url}/rest/v1/rpc/wavex_os_count_campaign_signups`, { + method: "POST", + headers: { + "Content-Type": "application/json", + apikey: key, + Authorization: `Bearer ${key}`, + }, + body: JSON.stringify({ p_utm_campaign: campaign }), + }); if (!res.ok) { return reply.status(502).send({ ok: false, error: `Supabase ${res.status}` }); } - const rows = (await res.json().catch(() => [])) as Array<{ user_id: string }>; - const distinct = new Set(rows.map((r) => r.user_id)).size; + const distinct = (await res.json().catch(() => 0)) as number; return reply.send({ ok: true, utm_campaign: campaign, new_auth_users: distinct }); }, ); diff --git a/packages/wavex-os-server/src/routes/booking-recovery.ts b/packages/wavex-os-server/src/routes/booking-recovery.ts new file mode 100644 index 0000000..28c71ac --- /dev/null +++ b/packages/wavex-os-server/src/routes/booking-recovery.ts @@ -0,0 +1,37 @@ +/** Abandoned booking recovery manual trigger (WAVAAAA-1200). + * + * POST /api/booking/recovery/run + * Board-only trigger that fires runAbandonedBookingRecoveryJob once. + * Safe: job is idempotent via the unique (intent_id) nudge log constraint. + * Useful for manually re-running after enabling WAVEX_ABANDONED_BOOKING_DRY_RUN=false. + */ + +import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify"; +import { assertBoard, AuthError } from "@wavex-os/auth-shim"; +import { runAbandonedBookingRecoveryJob } from "../jobs/abandoned-booking-recovery.js"; + +function authReq(req: FastifyRequest) { + return { method: req.method, headers: req.headers as Record }; +} + +export function registerBookingRecoveryRoute(app: FastifyInstance): void { + app.post( + "/api/booking/recovery/run", + async (req: FastifyRequest, reply: FastifyReply) => { + const ar = authReq(req); + try { + assertBoard(ar); + } catch (e) { + if (e instanceof AuthError) return reply.status(e.statusCode).send({ error: e.message }); + throw e; + } + try { + const result = await runAbandonedBookingRecoveryJob(); + return { ok: true, ...result }; + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + return reply.status(500).send({ ok: false, error: msg }); + } + }, + ); +} diff --git a/packages/wavex-os-server/src/routes/inference-allocation.ts b/packages/wavex-os-server/src/routes/inference-allocation.ts index 2944d5d..580a168 100644 --- a/packages/wavex-os-server/src/routes/inference-allocation.ts +++ b/packages/wavex-os-server/src/routes/inference-allocation.ts @@ -18,9 +18,14 @@ */ import { readFile, writeFile, mkdir } from "node:fs/promises"; import { join, dirname } from "node:path"; -import type { FastifyInstance } from "fastify"; +import type { FastifyInstance, FastifyRequest } from "fastify"; +import { assertBoard, AuthError } from "@wavex-os/auth-shim"; import { getWavexDataRoot } from "../state-bridge.js"; +function authReq(req: FastifyRequest) { + return { method: req.method, headers: req.headers as Record }; +} + const DEFAULT_SWARM_PCT = 70; export interface InferenceAllocation { @@ -65,6 +70,17 @@ export function registerInferenceAllocationRoute(app: FastifyInstance): void { }); app.put("/api/inference-allocation", async (req, reply) => { + // Board-only: changing the swarm/Pool-A split is an operator setting. An + // unauthenticated PUT with swarm_pct=100 would zero out Pool A and starve + // the onboarding wizard a new customer is sitting in front of. In dev mode + // the gate is a no-op (synthetic local board actor), so Mission Control + + // onboarding Pillar 2 keep working; in production it requires a board actor. + try { + assertBoard(authReq(req)); + } catch (e) { + if (e instanceof AuthError) return reply.status(e.statusCode).send({ ok: false, error: e.message }); + throw e; + } const body = (req.body ?? {}) as { swarm_pct?: unknown }; if (body.swarm_pct === undefined) { return reply.status(400).send({ ok: false, error: "swarm_pct is required" }); diff --git a/packages/wavex-os-server/src/routes/system-actions.ts b/packages/wavex-os-server/src/routes/system-actions.ts index 88db80c..7f79c5e 100644 --- a/packages/wavex-os-server/src/routes/system-actions.ts +++ b/packages/wavex-os-server/src/routes/system-actions.ts @@ -26,16 +26,41 @@ * Daemon is spawned non-blocking after each action so /api/system/health * returns the updated state within ~5s of the click. */ -import type { FastifyInstance } from "fastify"; +import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify"; import path from "node:path"; import { homedir, platform } from "node:os"; import { spawn, exec } from "node:child_process"; import { fileURLToPath } from "node:url"; import { promisify } from "node:util"; import { promises as fs } from "node:fs"; +import { assertBoard, AuthError } from "@wavex-os/auth-shim"; const execAsync = promisify(exec); +/** Mirror of the authReq helper used by activate.ts — projects the Fastify + * request into the minimal shape the auth-shim gates read. In dev mode the + * gate auto-populates a synthetic local board actor (no-op for the local + * operator); in production it requires a real board actor. */ +function authReq(req: FastifyRequest) { + return { method: req.method, headers: req.headers as Record }; +} + +/** Returns true and sends the error reply when the caller is NOT an + * authorized board actor. These handlers run git/launchctl/pnpm against the + * repo root, so an unauthenticated caller must never reach them. */ +function denyIfNotBoard(req: FastifyRequest, reply: FastifyReply): boolean { + try { + assertBoard(authReq(req)); + return false; + } catch (e) { + if (e instanceof AuthError) { + reply.status(e.statusCode).send({ ok: false, error: e.message }); + return true; + } + throw e; + } +} + const __dirname = path.dirname(fileURLToPath(import.meta.url)); // Walk up from packages/wavex-os-server/src/routes/ → repo root. const REPO_ROOT = path.resolve(__dirname, "../../../.."); @@ -58,7 +83,8 @@ async function spawnDaemonRefresh(): Promise { } export function registerSystemActionsRoute(app: FastifyInstance): void { - app.post("/api/system/discard-local-changes", async () => { + app.post("/api/system/discard-local-changes", async (req, reply) => { + if (denyIfNotBoard(req, reply)) return reply; try { const status = await execAsync("git status --porcelain", { cwd: REPO_ROOT }); if (!status.stdout.trim()) { @@ -96,7 +122,8 @@ export function registerSystemActionsRoute(app: FastifyInstance): void { } }); - app.post("/api/system/restart-processes", async () => { + app.post("/api/system/restart-processes", async (req, reply) => { + if (denyIfNotBoard(req, reply)) return reply; const labels = [ "com.wavex-os.mock-core", "com.wavex-os.wavex-os-server", @@ -133,7 +160,8 @@ export function registerSystemActionsRoute(app: FastifyInstance): void { return { ok: true, results, daemon_refreshed: refreshed }; }); - app.post("/api/system/pull-and-restart", async () => { + app.post("/api/system/pull-and-restart", async (req, reply) => { + if (denyIfNotBoard(req, reply)) return reply; let stashed = false; let stashLabel: string | null = null; try { diff --git a/scripts/resend-backfill.mjs b/scripts/resend-backfill.mjs new file mode 100755 index 0000000..2e35f4c --- /dev/null +++ b/scripts/resend-backfill.mjs @@ -0,0 +1,158 @@ +#!/usr/bin/env node +/** + * Resend audience backfill — WAVAAAA-1209 + * + * Fetches all wavex_os.auth_events rows where: + * resend_fired = false AND event_type = 'signup_confirmed' AND email IS NOT NULL + * then adds each as a Resend contact, then marks the row resend_fired=true. + * + * Usage: + * node scripts/resend-backfill.mjs [--dry-run] [--batch 100] + * + * Reads env from ~/.wavex-os/state/.env if RESEND_API_KEY is not already set. + */ + +import { readFileSync } from "fs"; +import { homedir } from "os"; +import { join } from "path"; + +// ─── env loader ──────────────────────────────────────────────────────────── + +function loadDotEnv(path) { + try { + const raw = readFileSync(path, "utf8"); + for (const line of raw.split("\n")) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith("#")) continue; + const idx = trimmed.indexOf("="); + if (idx < 0) continue; + const key = trimmed.slice(0, idx).trim(); + const val = trimmed.slice(idx + 1).trim(); + if (key && !(key in process.env)) process.env[key] = val; + } + } catch { + // ignore missing file + } +} + +loadDotEnv(join(homedir(), ".wavex-os", "state", ".env")); + +// ─── config ──────────────────────────────────────────────────────────────── + +const RESEND_API_KEY = process.env.RESEND_API_KEY; +const RESEND_AUDIENCE_ID = process.env.RESEND_AUDIENCE_ID; +const SUPABASE_URL = process.env.SUPABASE_URL; +const SUPABASE_KEY = process.env.SUPABASE_SERVICE_ROLE_KEY; + +const args = process.argv.slice(2); +const DRY_RUN = args.includes("--dry-run"); +const BATCH = (() => { + const i = args.indexOf("--batch"); + return i >= 0 && args[i + 1] ? parseInt(args[i + 1], 10) : 100; +})(); + +// ─── guards ──────────────────────────────────────────────────────────────── + +function fatal(msg) { console.error(`[backfill] FATAL: ${msg}`); process.exit(1); } + +if (!SUPABASE_URL) fatal("SUPABASE_URL not set"); +if (!SUPABASE_KEY) fatal("SUPABASE_SERVICE_ROLE_KEY not set"); +if (!RESEND_API_KEY && !DRY_RUN) fatal("RESEND_API_KEY not set (use --dry-run to preview without sending)"); +if (!RESEND_AUDIENCE_ID && !DRY_RUN) fatal("RESEND_AUDIENCE_ID not set (use --dry-run to preview without sending)"); + +// ─── Supabase helpers ────────────────────────────────────────────────────── + +async function rpc(funcName, params = {}) { + const res = await fetch(`${SUPABASE_URL}/rest/v1/rpc/${funcName}`, { + method: "POST", + headers: { + "Content-Type": "application/json", + apikey: SUPABASE_KEY, + Authorization: `Bearer ${SUPABASE_KEY}`, + }, + body: JSON.stringify(params), + }); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + throw new Error(`RPC ${funcName} failed: ${res.status} ${detail}`); + } + return res.json(); +} + +// ─── Resend helper ───────────────────────────────────────────────────────── + +async function addResendContact(email, utmCampaign) { + if (DRY_RUN) { + console.log(`[backfill] DRY-RUN: would add contact email=${email} utm=${utmCampaign ?? "(none)"}`); + return true; + } + + const body = { email, unsubscribed: false }; + if (utmCampaign) body.data = { utm_campaign: utmCampaign }; + + const ctrl = new AbortController(); + const t = setTimeout(() => ctrl.abort(), 15_000); + try { + const res = await fetch( + `https://api.resend.com/audiences/${encodeURIComponent(RESEND_AUDIENCE_ID)}/contacts`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${RESEND_API_KEY}`, + }, + body: JSON.stringify(body), + signal: ctrl.signal, + } + ); + clearTimeout(t); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + console.error(`[backfill] Resend error for ${email}: ${res.status} ${detail}`); + return false; + } + return true; + } catch (e) { + clearTimeout(t); + console.error(`[backfill] Resend fetch error for ${email}: ${e.message}`); + return false; + } +} + +// ─── main ────────────────────────────────────────────────────────────────── + +async function main() { + console.log(`[backfill] Starting${DRY_RUN ? " (DRY-RUN)" : ""} batch=${BATCH}`); + + let total = 0, succeeded = 0, failed = 0; + + while (true) { + const rows = await rpc("wavex_os_get_unfired_auth_events", { p_limit: BATCH }); + if (!rows.length) break; + + console.log(`[backfill] Processing ${rows.length} row(s)...`); + + for (const row of rows) { + const ok = await addResendContact(row.email, row.utm_campaign); + total++; + if (ok) { + if (!DRY_RUN) { + await rpc("wavex_os_mark_auth_event_fired", { p_id: row.id }); + } + console.log(`[backfill] ✓ ${row.email} (id=${row.id})`); + succeeded++; + } else { + console.warn(`[backfill] ✗ ${row.email} (id=${row.id}) — will retry on next run`); + failed++; + } + } + + // If the batch wasn't full, we've exhausted the queue + if (rows.length < BATCH) break; + } + + console.log(`[backfill] Done — total=${total} succeeded=${succeeded} failed=${failed}`); + if (failed > 0) process.exit(1); +} + +main().catch((err) => { console.error("[backfill] Unexpected error:", err); process.exit(1); }); diff --git a/supabase/functions/create-booking-checkout-session/index.ts b/supabase/functions/create-booking-checkout-session/index.ts new file mode 100644 index 0000000..51c96d0 --- /dev/null +++ b/supabase/functions/create-booking-checkout-session/index.ts @@ -0,0 +1,162 @@ +/** + * POST /functions/v1/create-booking-checkout-session + * + * Body: { booking_intent_id: string, success_url: string, cancel_url: string } + * Auth: Supabase JWT in Authorization header (user must own the intent or be service_role). + * + * Returns: { url: string } — Stripe Checkout session URL. + * + * Flow: + * 1. Verify JWT and resolve user_id. + * 2. Load the booking_intent row (verify it belongs to the caller). + * 3. Create a Stripe Checkout session (mode: payment, one-time). + * 4. Call wavex_os_set_booking_intent_pending_payment() to freeze the + * 30-min auto-cancel clock. + * 5. Return the Stripe URL. + * + * Deploy: + * supabase functions deploy create-booking-checkout-session + * + * Env vars: + * STRIPE_SECRET_KEY_TEST_ENV — preferred (test mode); falls back to STRIPE_SECRET_KEY + * SUPABASE_URL — auto-injected + * SUPABASE_SERVICE_ROLE_KEY — auto-injected + * SUPABASE_ANON_KEY — auto-injected + */ +// @ts-expect-error — Deno-style import resolved at runtime +import Stripe from "https://esm.sh/stripe@17.5.0?target=denonext"; +// @ts-expect-error — Deno-style import +import { createClient } from "https://esm.sh/@supabase/supabase-js@2.46.1"; + +const stripeKey = + Deno.env.get("STRIPE_SECRET_KEY_TEST_ENV") ?? Deno.env.get("STRIPE_SECRET_KEY")!; +const supabaseUrl = Deno.env.get("SUPABASE_URL")!; +const serviceKey = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!; +const anonKey = Deno.env.get("SUPABASE_ANON_KEY")!; + +const stripe = new Stripe(stripeKey, { + apiVersion: "2025-09-30.clover" as Stripe.StripeConfig["apiVersion"], + httpClient: Stripe.createFetchHttpClient(), +}); + +const sb = createClient(supabaseUrl, serviceKey, { + auth: { persistSession: false, autoRefreshToken: false }, +}); + +const corsHeaders = { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Headers": "authorization, x-client-info, apikey, content-type", + "Access-Control-Allow-Methods": "POST, OPTIONS", +}; + +function json(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json", ...corsHeaders }, + }); +} + +Deno.serve(async (req: Request) => { + if (req.method === "OPTIONS") return new Response(null, { headers: corsHeaders }); + if (req.method !== "POST") return json({ error: "method_not_allowed" }, 405); + + const authHeader = req.headers.get("Authorization"); + if (!authHeader) return json({ error: "unauthorized" }, 401); + + // Verify the JWT by hitting auth.getUser as the caller. + const userClient = createClient(supabaseUrl, anonKey, { + global: { headers: { Authorization: authHeader } }, + auth: { persistSession: false, autoRefreshToken: false }, + }); + const { data: userData, error: userErr } = await userClient.auth.getUser(); + if (userErr || !userData.user) return json({ error: "unauthorized" }, 401); + const userId = userData.user.id; + const email = userData.user.email ?? undefined; + + let body: { booking_intent_id?: string; success_url?: string; cancel_url?: string }; + try { + body = await req.json(); + } catch { + return json({ error: "invalid_json" }, 400); + } + + const { booking_intent_id, success_url, cancel_url } = body; + if (!booking_intent_id || !success_url || !cancel_url) { + return json({ error: "missing_fields" }, 400); + } + + // Load the booking intent via RPC (wavex_os schema is not in PostgREST exposed list; + // all wavex_os table access goes through public-schema security-definer RPCs). + const { data: intent, error: intentErr } = await sb.rpc("wavex_os_get_booking_intent", { + p_intent_id: booking_intent_id, + }); + + if (intentErr || !intent) return json({ error: "intent_not_found" }, 404); + if (intent.user_id && intent.user_id !== userId) return json({ error: "forbidden" }, 403); + if (intent.status === "confirmed") return json({ error: "already_confirmed" }, 409); + if (intent.status === "cancelled") return json({ error: "intent_cancelled" }, 410); + + // If already in pending_payment with an existing session, reuse it. + if (intent.status === "pending_payment" && intent.stripe_checkout_session_id) { + const existing = await stripe.checkout.sessions.retrieve(intent.stripe_checkout_session_id); + if (existing.status === "open" && existing.url) { + return json({ url: existing.url }); + } + // Session expired or completed — fall through to create a fresh one. + } + + // Look up or create Stripe customer. + const customers = await stripe.customers.list({ email, limit: 1 }); + let customerId = customers.data[0]?.id; + if (!customerId) { + const newCustomer = await stripe.customers.create({ + email, + metadata: { supabase_user_id: userId }, + }); + customerId = newCustomer.id; + } + + const session = await stripe.checkout.sessions.create({ + customer: customerId, + mode: "payment", + line_items: [ + { + price_data: { + currency: intent.currency ?? "usd", + unit_amount: intent.experience_price_cents, + product_data: { + name: intent.experience_name ?? "Experience Booking", + }, + }, + quantity: 1, + }, + ], + metadata: { + booking_intent_id, + supabase_user_id: userId, + }, + client_reference_id: userId, + success_url: success_url + (success_url.includes("?") ? "&" : "?") + "session_id={CHECKOUT_SESSION_ID}", + cancel_url, + }); + + // Freeze the 30-min auto-cancel clock: pending → pending_payment. + const { data: rpcData, error: rpcErr } = await sb.rpc( + "wavex_os_set_booking_intent_pending_payment", + { + p_intent_id: booking_intent_id, + p_stripe_checkout_session_id: session.id, + }, + ); + if (rpcErr) { + console.error("wavex_os_set_booking_intent_pending_payment failed", rpcErr); + // Non-fatal: Stripe session is created; caller can still redirect. + // Log but don't block the response. + } + const rpcResult = rpcData as { ok?: boolean; reason?: string } | null; + if (rpcResult && !rpcResult.ok) { + console.warn("booking intent status transition failed", rpcResult.reason); + } + + return json({ url: session.url }); +}); diff --git a/supabase/functions/wavex-os-booking-webhook/index.ts b/supabase/functions/wavex-os-booking-webhook/index.ts new file mode 100644 index 0000000..da8fbd2 --- /dev/null +++ b/supabase/functions/wavex-os-booking-webhook/index.ts @@ -0,0 +1,179 @@ +/** + * Stripe webhook → booking_intents + public.bookings + * + * Handles Stripe events for one-time experience booking payments: + * - checkout.session.completed → confirm intent, create bookings row, + * emit booking_confirmed Telegram event + * - checkout.session.expired → revert pending_payment → pending so + * the intent becomes auto-cancel-eligible + * + * This handler only acts on sessions that have metadata.booking_intent_id set. + * Sessions without that key (e.g. subscription checkouts) are silently ignored. + * + * Deploy: + * supabase functions deploy wavex-os-booking-webhook --no-verify-jwt + * + * Env vars: + * STRIPE_SECRET_KEY_TEST_ENV — preferred; falls back to STRIPE_SECRET_KEY + * WAVEX_OS_BOOKING_WEBHOOK_SECRET — Stripe signing secret for THIS endpoint + * SUPABASE_URL — auto-injected + * SUPABASE_SERVICE_ROLE_KEY — auto-injected + * + * Stripe dashboard configuration: + * Endpoint URL: https://.supabase.co/functions/v1/wavex-os-booking-webhook + * Events: checkout.session.completed, checkout.session.expired + */ +// @ts-expect-error — Deno-style import resolved at runtime +import Stripe from "https://esm.sh/stripe@17.5.0?target=denonext"; +// @ts-expect-error — Deno-style import +import { createClient } from "https://esm.sh/@supabase/supabase-js@2.46.1"; + +const stripeKey = + Deno.env.get("STRIPE_SECRET_KEY_TEST_ENV") ?? Deno.env.get("STRIPE_SECRET_KEY")!; +const webhookSecret = Deno.env.get("WAVEX_OS_BOOKING_WEBHOOK_SECRET")!; +const supabaseUrl = Deno.env.get("SUPABASE_URL")!; +const serviceKey = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!; + +const stripe = new Stripe(stripeKey, { + apiVersion: "2025-09-30.clover" as Stripe.StripeConfig["apiVersion"], + httpClient: Stripe.createFetchHttpClient(), +}); +const cryptoProvider = Stripe.createSubtleCryptoProvider(); + +const sb = createClient(supabaseUrl, serviceKey, { + auth: { persistSession: false, autoRefreshToken: false }, +}); + +async function handleCheckoutCompleted(session: Stripe.Checkout.Session): Promise { + const bookingIntentId = session.metadata?.booking_intent_id; + if (!bookingIntentId) return; // Not a booking session — skip. + + const paymentIntentId = + typeof session.payment_intent === "string" + ? session.payment_intent + : session.payment_intent?.id ?? null; + + // Atomically confirm the intent and create the public.bookings row. + const { data: confirmData, error: confirmErr } = await sb.rpc( + "wavex_os_confirm_booking_intent", + { + p_intent_id: bookingIntentId, + p_stripe_checkout_session_id: session.id, + p_stripe_payment_intent_id: paymentIntentId, + }, + ); + if (confirmErr) { + console.error("wavex_os_confirm_booking_intent failed", confirmErr); + throw confirmErr; + } + + const result = confirmData as { ok?: boolean; reason?: string; booking_id?: string; idempotent?: boolean } | null; + if (!result?.ok) { + console.error("confirm_booking_intent returned not-ok", result); + return; + } + if (result.idempotent) { + console.info(`booking_intent ${bookingIntentId} already confirmed (idempotent replay)`); + return; + } + + // Emit booking_confirmed Telegram event. + const userId = session.client_reference_id ?? session.metadata?.supabase_user_id ?? null; + if (userId && result.booking_id) { + const { error: tgErr } = await sb.rpc("wavex_os_emit_booking_confirmed_for_user", { + p_booking_id: result.booking_id, + p_user_id: userId, + }); + if (tgErr) { + // Non-fatal: booking is confirmed; telegram notification failure should not + // roll back the payment acknowledgment. + console.error( + `emit booking_confirmed failed booking=${result.booking_id}`, + tgErr, + ); + } + } + + console.info( + `booking confirmed: intent=${bookingIntentId} booking=${result.booking_id} session=${session.id}`, + ); +} + +async function handleCheckoutExpired(session: Stripe.Checkout.Session): Promise { + const bookingIntentId = session.metadata?.booking_intent_id; + if (!bookingIntentId) return; + + // Revert to pending so the 30-min auto-cancel clock can fire. + const { error } = await sb + .schema("wavex_os") + .from("booking_intents") + .update({ status: "pending", updated_at: new Date().toISOString() }) + .eq("id", bookingIntentId) + .eq("status", "pending_payment"); // Only revert if still in the redirect state. + + if (error) { + console.error(`revert pending_payment failed intent=${bookingIntentId}`, error); + } else { + console.info(`checkout expired → reverted to pending intent=${bookingIntentId}`); + } +} + +Deno.serve(async (req: Request) => { + if (req.method !== "POST") { + return new Response("Method not allowed", { status: 405 }); + } + const sig = req.headers.get("stripe-signature"); + if (!sig) return new Response("Missing signature", { status: 400 }); + + const body = await req.text(); + let event: Stripe.Event; + try { + event = await stripe.webhooks.constructEventAsync(body, sig, webhookSecret, undefined, cryptoProvider); + } catch (err) { + console.error("signature verification failed", err); + return new Response(`Webhook Error: ${(err as Error).message}`, { status: 400 }); + } + + // Idempotency guard reusing the existing stripe_webhook_events table. + const { data: idemData, error: idemErr } = await sb.rpc("wavex_os_record_webhook_event", { + p_id: event.id, + p_type: event.type, + p_api_version: event.api_version, + p_payload: event as unknown as Record, + }); + if (idemErr) { + console.error("idempotency insert failed", idemErr); + return new Response("Internal error", { status: 500 }); + } + const isDup = Array.isArray(idemData) ? idemData[0]?.is_duplicate : idemData?.is_duplicate; + if (isDup) { + return new Response(JSON.stringify({ ok: true, duplicate: true }), { + headers: { "Content-Type": "application/json" }, + }); + } + + try { + switch (event.type) { + case "checkout.session.completed": + await handleCheckoutCompleted(event.data.object as Stripe.Checkout.Session); + break; + case "checkout.session.expired": + await handleCheckoutExpired(event.data.object as Stripe.Checkout.Session); + break; + default: + // Unhandled event type — idempotency row recorded, ack to Stripe. + break; + } + + return new Response(JSON.stringify({ ok: true }), { + headers: { "Content-Type": "application/json" }, + }); + } catch (err) { + await sb.rpc("wavex_os_mark_webhook_event_error", { + p_id: event.id, + p_error: (err as Error).message, + }); + console.error("event handler failed", err); + return new Response("Handler error", { status: 500 }); + } +}); diff --git a/supabase/migrations/20260603000002_wavex_os_booking_intents.sql b/supabase/migrations/20260603000002_wavex_os_booking_intents.sql new file mode 100644 index 0000000..79086c6 --- /dev/null +++ b/supabase/migrations/20260603000002_wavex_os_booking_intents.sql @@ -0,0 +1,253 @@ +-- WaveX OS — booking_intents table + Stripe checkout confirmation RPC (WAVAAAA-1195) +-- +-- Problem: Telegram bot creates booking_intent telegram events but there is +-- no payment path to transition them. The 30-min auto-cancel fires on every +-- unconfirmed intent because there is no way to pause the clock while the +-- user is in the Stripe redirect. +-- +-- This migration adds: +-- 1. wavex_os.booking_intents — staging table for the intent-to-pay +-- 2. wavex_os_confirm_booking_intent() — atomic confirm: update intent + +-- insert public.bookings row (service_role RPC) +-- 3. wavex_os_cancel_stale_booking_intents() — auto-cancel only 'pending' +-- (not 'pending_payment') intents older +-- than 30 min. Called by a pg_cron job +-- or the booking-cleanup Supabase Function. +-- 4. wavex_os_set_booking_intent_pending_payment() — update status to +-- 'pending_payment' when user is sent to +-- Stripe; takes the checkout session id. + +begin; + +-- ─── booking_intents table ──────────────────────────────────────────────────── + +create table if not exists wavex_os.booking_intents ( + id uuid primary key default gen_random_uuid(), + user_id uuid null, -- auth.users.id (if linked) + telegram_user_id text null, -- Telegram bot user + experience_id text null, -- experience / product slug + experience_name text null, + experience_price_cents bigint not null, -- charge amount, in cents + currency text not null default 'usd', + booking_time timestamptz null, -- requested time slot + stripe_checkout_session_id text null, -- Stripe checkout.session id + stripe_payment_intent_id text null, -- Stripe PaymentIntent id (set on confirm) + status text not null default 'pending', + cancelled_at timestamptz null, + confirmed_at timestamptz null, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + constraint booking_intents_status_check + check (status in ('pending', 'pending_payment', 'confirmed', 'cancelled')) +); + +create index if not exists booking_intents_user_id_idx + on wavex_os.booking_intents (user_id) + where user_id is not null; + +create index if not exists booking_intents_status_created_idx + on wavex_os.booking_intents (status, created_at) + where status in ('pending', 'pending_payment'); + +create unique index if not exists booking_intents_checkout_session_uniq + on wavex_os.booking_intents (stripe_checkout_session_id) + where stripe_checkout_session_id is not null; + +comment on table wavex_os.booking_intents is + 'Staging table for experience booking intents. Created by the Telegram bot ' + 'when a user requests a booking. Transitions: pending → pending_payment ' + '(on Stripe redirect) → confirmed (on checkout.session.completed webhook). ' + 'Confirmed intents produce a public.bookings row via wavex_os_confirm_booking_intent(). ' + 'Only pending intents (not pending_payment) are auto-cancelled by the 30-min cleanup.'; + +-- ─── set pending_payment (called by create-booking-checkout-session function) ─ + +create or replace function public.wavex_os_set_booking_intent_pending_payment( + p_intent_id uuid, + p_stripe_checkout_session_id text +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_rows int; +begin + update wavex_os.booking_intents + set status = 'pending_payment', + stripe_checkout_session_id = p_stripe_checkout_session_id, + updated_at = now() + where id = p_intent_id + and status = 'pending'; + + get diagnostics v_rows = row_count; + if v_rows = 0 then + -- Already in another terminal state or doesn't exist — surface to caller. + return jsonb_build_object('ok', false, 'reason', 'not_pending_or_not_found'); + end if; + return jsonb_build_object('ok', true); +end; +$$; + +revoke all on function public.wavex_os_set_booking_intent_pending_payment(uuid, text) from public; +grant execute on function public.wavex_os_set_booking_intent_pending_payment(uuid, text) to service_role; + +comment on function public.wavex_os_set_booking_intent_pending_payment is + 'Atomically transition a booking_intent from pending → pending_payment and record ' + 'the Stripe checkout session id. Called by the create-booking-checkout-session edge ' + 'function immediately before redirecting the user to Stripe. Intents in ' + 'pending_payment are NOT auto-cancelled by the 30-min cleanup job.'; + +-- ─── confirm booking intent (called by booking webhook on payment success) ──── + +create or replace function public.wavex_os_confirm_booking_intent( + p_intent_id uuid, + p_stripe_checkout_session_id text, + p_stripe_payment_intent_id text default null +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_intent wavex_os.booking_intents%rowtype; + v_booking_id uuid; +begin + -- Idempotency: if already confirmed, return the existing booking_id from + -- the public.bookings row tied to this checkout session. + select * into v_intent + from wavex_os.booking_intents + where id = p_intent_id + limit 1; + + if not found then + return jsonb_build_object('ok', false, 'reason', 'intent_not_found'); + end if; + + if v_intent.status = 'confirmed' then + -- Already confirmed; look up the booking row and return idempotently. + select id into v_booking_id + from public.bookings + where booking_intent_id = p_intent_id + limit 1; + return jsonb_build_object('ok', true, 'idempotent', true, 'booking_id', v_booking_id); + end if; + + if v_intent.status not in ('pending', 'pending_payment') then + return jsonb_build_object('ok', false, 'reason', 'intent_not_payable', 'status', v_intent.status); + end if; + + -- Confirm the intent. + update wavex_os.booking_intents + set status = 'confirmed', + stripe_checkout_session_id = p_stripe_checkout_session_id, + stripe_payment_intent_id = p_stripe_payment_intent_id, + confirmed_at = now(), + updated_at = now() + where id = p_intent_id; + + -- Insert the public.bookings row. + insert into public.bookings ( + user_id, + booking_intent_id, + booking_status, + amount, + currency, + booking_time, + experience_id, + experience_name, + stripe_checkout_session_id, + stripe_payment_intent_id, + paid_at + ) values ( + v_intent.user_id, + p_intent_id, + 'confirmed', + (v_intent.experience_price_cents / 100.0)::numeric, + upper(v_intent.currency), + v_intent.booking_time, + v_intent.experience_id, + v_intent.experience_name, + p_stripe_checkout_session_id, + p_stripe_payment_intent_id, + now() + ) + returning id into v_booking_id; + + return jsonb_build_object('ok', true, 'booking_id', v_booking_id); +end; +$$; + +revoke all on function public.wavex_os_confirm_booking_intent(uuid, text, text) from public; +grant execute on function public.wavex_os_confirm_booking_intent(uuid, text, text) to service_role; + +comment on function public.wavex_os_confirm_booking_intent is + 'Atomically confirm a booking_intent and create the public.bookings row. ' + 'Called by the wavex-os-booking-webhook edge function on checkout.session.completed. ' + 'Idempotent: replaying the same p_intent_id returns ok=true with idempotent=true. ' + 'Only accepts intents in pending or pending_payment state.'; + +-- ─── cancel stale pending intents (30-min clock — NOT pending_payment) ─────── + +create or replace function public.wavex_os_cancel_stale_booking_intents( + p_older_than_minutes int default 30 +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_cancelled_ids uuid[]; +begin + update wavex_os.booking_intents + set status = 'cancelled', + cancelled_at = now(), + updated_at = now() + where status = 'pending' -- never cancel pending_payment + and created_at < now() - (p_older_than_minutes || ' minutes')::interval + returning id + into v_cancelled_ids; + + return jsonb_build_object('cancelled', coalesce(array_length(v_cancelled_ids, 1), 0)); +end; +$$; + +revoke all on function public.wavex_os_cancel_stale_booking_intents(int) from public; +grant execute on function public.wavex_os_cancel_stale_booking_intents(int) to service_role; + +comment on function public.wavex_os_cancel_stale_booking_intents is + 'Cancel booking_intents that remain in ''pending'' status for longer than ' + 'p_older_than_minutes (default 30). Intentionally does NOT cancel ' + 'pending_payment intents — those are live Stripe redirect sessions and ' + 'will be resolved by the checkout.session.completed or checkout.session.expired webhook.'; + +-- ─── extend public.bookings to accept booking_intent_id ────────────────────── + +alter table public.bookings + add column if not exists booking_intent_id uuid null + references wavex_os.booking_intents(id) on delete set null; + +alter table public.bookings + add column if not exists experience_id text null; + +alter table public.bookings + add column if not exists experience_name text null; + +alter table public.bookings + add column if not exists stripe_checkout_session_id text null; + +alter table public.bookings + add column if not exists stripe_payment_intent_id text null; + +create unique index if not exists bookings_booking_intent_id_uniq + on public.bookings (booking_intent_id) + where booking_intent_id is not null; + +create unique index if not exists bookings_stripe_checkout_session_uniq + on public.bookings (stripe_checkout_session_id) + where stripe_checkout_session_id is not null; + +commit; diff --git a/supabase/migrations/20260603000003_wavex_os_abandoned_booking_recovery.sql b/supabase/migrations/20260603000003_wavex_os_abandoned_booking_recovery.sql new file mode 100644 index 0000000..9150394 --- /dev/null +++ b/supabase/migrations/20260603000003_wavex_os_abandoned_booking_recovery.sql @@ -0,0 +1,174 @@ +-- WaveX OS — abandoned booking recovery nudges (WAVAAAA-1197) +-- +-- Users whose booking_intents were auto-cancelled before the Stripe checkout +-- flow was added (bfeff1f5) never had a way to pay. This migration gives us +-- the infrastructure to identify those users and send a one-time recovery +-- nudge via the Telegram concierge bot. +-- +-- This migration adds: +-- • wavex_os.abandoned_booking_recovery_nudges — append-only nudge log +-- • wavex_os_list_abandoned_booking_candidates() — find cancelled intents +-- that have a resolvable Telegram chat_id and haven't been nudged yet +-- • wavex_os_record_abandoned_booking_nudge() — record a nudge send +-- +-- The job in packages/wavex-os-server/src/jobs/abandoned-booking-recovery.ts +-- calls these RPCs. It defaults to dry-run (WAVEX_ABANDONED_BOOKING_DRY_RUN=true) +-- so the recoverable count is logged on every startup without sending anything. +-- Set WAVEX_ABANDONED_BOOKING_DRY_RUN=false to actually send once CEO approves. + +begin; + +-- ─── nudge log ──────────────────────────────────────────────────────────────── + +create table if not exists wavex_os.abandoned_booking_recovery_nudges ( + id uuid primary key default gen_random_uuid(), + intent_id uuid not null + references wavex_os.booking_intents(id) + on delete cascade, + telegram_user_id text not null, + chat_id text not null, + message_text text not null, + message_id text, + dry_run boolean not null default true, + sent_at timestamptz not null default now(), + created_at timestamptz not null default now(), + constraint abandoned_booking_recovery_nudges_intent_uniq unique (intent_id) +); + +comment on table wavex_os.abandoned_booking_recovery_nudges is + 'Append-only log of one-time Telegram recovery nudges sent to users whose ' + 'booking_intents were cancelled before the Stripe checkout flow existed. ' + 'Unique on intent_id so a replay cannot double-send.'; + +-- ─── list candidates ────────────────────────────────────────────────────────── +-- Returns all cancelled booking_intents where: +-- 1. A telegram chat_id can be resolved (by telegram_user_id or user_id) +-- 2. The user does not already have a confirmed booking (they retried on own) +-- 3. The intent has not already been nudged (idempotent via nudge log) +-- +-- chat_id resolution order: +-- a. Most recent telegram_event matching booking_intent.telegram_user_id +-- b. Most recent telegram_event matching booking_intent.user_id (auth link) + +create or replace function public.wavex_os_list_abandoned_booking_candidates() +returns table ( + intent_id uuid, + telegram_user_id text, + chat_id text, + experience_name text, + experience_price_cents bigint, + currency text, + created_at timestamptz, + cancelled_at timestamptz +) +language sql +stable +security definer +set search_path = wavex_os, public +as $$ + select + bi.id, + coalesce(bi.telegram_user_id, te_uid.telegram_user_id) as telegram_user_id, + coalesce(te_tgid.chat_id, te_uid.chat_id) as chat_id, + bi.experience_name, + bi.experience_price_cents, + bi.currency, + bi.created_at, + bi.cancelled_at + from wavex_os.booking_intents bi + + -- Primary: resolve chat_id by telegram_user_id + left join lateral ( + select te.chat_id + from wavex_os.telegram_events te + where te.telegram_user_id = bi.telegram_user_id + and te.chat_id is not null + order by te.occurred_at desc + limit 1 + ) te_tgid on bi.telegram_user_id is not null + + -- Fallback: resolve chat_id by auth user_id + left join lateral ( + select te.telegram_user_id, te.chat_id + from wavex_os.telegram_events te + where te.user_id = bi.user_id + and te.chat_id is not null + order by te.occurred_at desc + limit 1 + ) te_uid on bi.user_id is not null + and te_tgid.chat_id is null + + where bi.status = 'cancelled' + -- Only include rows where we have a destination + and coalesce(te_tgid.chat_id, te_uid.chat_id) is not null + -- Skip users who already confirmed a booking (converted without nudge) + and not exists ( + select 1 from wavex_os.booking_intents bi2 + where bi2.status = 'confirmed' + and ( + (bi.telegram_user_id is not null + and bi2.telegram_user_id = bi.telegram_user_id) + or + (bi.user_id is not null + and bi2.user_id = bi.user_id) + ) + ) + -- Skip if already nudged (idempotent) + and not exists ( + select 1 from wavex_os.abandoned_booking_recovery_nudges n + where n.intent_id = bi.id + ) + + order by bi.created_at asc +$$; + +revoke all on function public.wavex_os_list_abandoned_booking_candidates() from public; +grant execute on function public.wavex_os_list_abandoned_booking_candidates() to service_role; + +comment on function public.wavex_os_list_abandoned_booking_candidates is + 'Return cancelled booking_intents that are recoverable via a Telegram nudge. ' + 'Resolves chat_id from wavex_os.telegram_events (by telegram_user_id, then user_id). ' + 'Excludes: already-confirmed users, already-nudged intents. ' + 'Called by the abandoned-booking-recovery server job on startup.'; + +-- ─── record nudge ───────────────────────────────────────────────────────────── +-- Idempotent: ON CONFLICT DO NOTHING on the unique (intent_id) constraint. +-- A replay (e.g. server restart with DRY_RUN=true after a live run) never +-- double-counts or overwrites an already-sent row. + +create or replace function public.wavex_os_record_abandoned_booking_nudge( + p_intent_id uuid, + p_telegram_user_id text, + p_chat_id text, + p_message_text text, + p_message_id text default null, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_id uuid; +begin + insert into wavex_os.abandoned_booking_recovery_nudges ( + intent_id, telegram_user_id, chat_id, message_text, message_id, dry_run + ) values ( + p_intent_id, p_telegram_user_id, p_chat_id, p_message_text, p_message_id, p_dry_run + ) + on conflict (intent_id) do nothing + returning id into v_id; + return jsonb_build_object('id', v_id); +end; +$$; + +revoke all on function public.wavex_os_record_abandoned_booking_nudge(uuid, text, text, text, text, boolean) from public; +grant execute on function public.wavex_os_record_abandoned_booking_nudge(uuid, text, text, text, text, boolean) to service_role; + +comment on function public.wavex_os_record_abandoned_booking_nudge is + 'Log a recovery nudge send (real or dry-run) into ' + 'wavex_os.abandoned_booking_recovery_nudges. ON CONFLICT DO NOTHING on ' + 'intent_id prevents double-logging on server restarts.'; + +commit; diff --git a/supabase/migrations/20260603000003_wavex_os_get_booking_intent_rpc.sql b/supabase/migrations/20260603000003_wavex_os_get_booking_intent_rpc.sql new file mode 100644 index 0000000..7a38f31 --- /dev/null +++ b/supabase/migrations/20260603000003_wavex_os_get_booking_intent_rpc.sql @@ -0,0 +1,41 @@ +-- wavex_os_get_booking_intent RPC (WAVAAAA-1198 smoke-test fix) +-- +-- The wavex_os schema is not in the PostgREST exposed schema list, so +-- .schema("wavex_os").from("booking_intents") in the edge function returns +-- intent_not_found. Pattern in this codebase: use public-schema RPCs with +-- security definer + search_path to proxy access to wavex_os tables. + +create or replace function public.wavex_os_get_booking_intent( + p_intent_id uuid +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_row wavex_os.booking_intents%rowtype; +begin + select * into v_row + from wavex_os.booking_intents + where id = p_intent_id + limit 1; + + if not found then + return null; + end if; + + return jsonb_build_object( + 'id', v_row.id, + 'user_id', v_row.user_id, + 'experience_name', v_row.experience_name, + 'experience_price_cents', v_row.experience_price_cents, + 'currency', v_row.currency, + 'status', v_row.status, + 'stripe_checkout_session_id', v_row.stripe_checkout_session_id + ); +end; +$$; + +revoke all on function public.wavex_os_get_booking_intent(uuid) from public; +grant execute on function public.wavex_os_get_booking_intent(uuid) to service_role; diff --git a/supabase/migrations/20260603000004_wavex_os_create_booking_intent_rpc.sql b/supabase/migrations/20260603000004_wavex_os_create_booking_intent_rpc.sql new file mode 100644 index 0000000..d81d4af --- /dev/null +++ b/supabase/migrations/20260603000004_wavex_os_create_booking_intent_rpc.sql @@ -0,0 +1,58 @@ +-- wavex_os_create_booking_intent RPC (WAVAAAA-1203) +-- +-- The Telegram bot needs a way to create rows in wavex_os.booking_intents. +-- The wavex_os schema is not exposed to PostgREST, so direct table access +-- from the bot (or any edge function without Deno.env.SUPABASE_SERVICE_ROLE_KEY +-- + schema override) fails with 'relation not found'. +-- +-- This RPC gives the bot a stable public-schema entry point, following the +-- same security-definer + search_path pattern as the other booking RPCs. + +create or replace function public.wavex_os_create_booking_intent( + p_telegram_user_id text, + p_experience_id text, + p_experience_name text, + p_experience_price_cents bigint, + p_currency text default 'usd', + p_booking_time timestamptz default null, + p_user_id uuid default null +) +returns jsonb +language plpgsql +security definer +set search_path = wavex_os, public +as $$ +declare + v_id uuid; +begin + insert into wavex_os.booking_intents ( + telegram_user_id, + user_id, + experience_id, + experience_name, + experience_price_cents, + currency, + booking_time + ) values ( + p_telegram_user_id, + p_user_id, + p_experience_id, + p_experience_name, + p_experience_price_cents, + lower(p_currency), + p_booking_time + ) + returning id into v_id; + + return jsonb_build_object('id', v_id); +end; +$$; + +revoke all on function public.wavex_os_create_booking_intent(text, text, text, bigint, text, timestamptz, uuid) from public; +grant execute on function public.wavex_os_create_booking_intent(text, text, text, bigint, text, timestamptz, uuid) to service_role; + +comment on function public.wavex_os_create_booking_intent is + 'Create a row in wavex_os.booking_intents from the Telegram bot or any ' + 'service_role caller. The wavex_os schema is not PostgREST-exposed so this ' + 'public security-definer wrapper is the correct entry point. Returns {id} ' + 'for use in the create-booking-checkout-session call.'; diff --git a/supabase/migrations/20260603000005_wavex_os_auth_events_rpcs.sql b/supabase/migrations/20260603000005_wavex_os_auth_events_rpcs.sql new file mode 100644 index 0000000..17cadcd --- /dev/null +++ b/supabase/migrations/20260603000005_wavex_os_auth_events_rpcs.sql @@ -0,0 +1,115 @@ +-- WaveX OS — auth_events RPCs +-- +-- The wavex_os schema is NOT exposed via PostgREST (intentional — same +-- pattern as wavex_os_client_rpcs). These SECURITY DEFINER functions in +-- the public schema are the only write/read path for auth_events from +-- server-side code using the service-role key. +-- +-- Three functions: +-- wavex_os_record_auth_event — upsert one signup event row +-- wavex_os_get_unfired_auth_events — read unfired signup_confirmed rows (backfill) +-- wavex_os_mark_auth_event_fired — mark one row resend_fired=true (backfill) + +-- ─── 1. Record an auth event ───────────────────────────────────────────────── +-- Called from wavex-os-server POST /api/auth-events. Inserts the event row +-- and returns the new row id (or null if duplicate-suppressed). +create or replace function public.wavex_os_record_auth_event( + p_user_id text, + p_email text, + p_event_type text, + p_utm_campaign text, + p_utm_source text, + p_ref text, + p_resend_fired boolean +) +returns uuid +language plpgsql +security definer +set search_path = '' +as $$ +declare + v_id uuid; +begin + insert into wavex_os.auth_events ( + user_id, email, event_type, utm_campaign, utm_source, ref, resend_fired + ) + values ( + p_user_id, p_email, p_event_type, p_utm_campaign, p_utm_source, p_ref, p_resend_fired + ) + on conflict do nothing + returning id into v_id; + + return v_id; +end; +$$; + +-- Service-role key is used server-side; no user session involved. +grant execute on function public.wavex_os_record_auth_event(text,text,text,text,text,text,boolean) + to service_role; + +-- ─── 2. Get unfired signup_confirmed events for backfill ───────────────────── +-- Returns up to p_limit rows where resend_fired=false, event_type='signup_confirmed', +-- and email is not null. Ordered oldest-first for deterministic backfill. +create or replace function public.wavex_os_get_unfired_auth_events( + p_limit int default 100 +) +returns table ( + id uuid, + email text, + utm_campaign text, + created_at timestamptz +) +language sql +security definer +set search_path = '' +as $$ + select id, email, utm_campaign, created_at + from wavex_os.auth_events + where resend_fired = false + and event_type = 'signup_confirmed' + and email is not null + order by created_at asc + limit greatest(1, least(p_limit, 1000)); +$$; + +grant execute on function public.wavex_os_get_unfired_auth_events(int) + to service_role; + +-- ─── 3. Mark one event as resend_fired ─────────────────────────────────────── +-- Called once per row during backfill after Resend contact is created. +create or replace function public.wavex_os_mark_auth_event_fired( + p_id uuid +) +returns void +language sql +security definer +set search_path = '' +as $$ + update wavex_os.auth_events + set resend_fired = true + where id = p_id; +$$; + +grant execute on function public.wavex_os_mark_auth_event_fired(uuid) + to service_role; + +-- ─── 4. Count distinct signup_confirmed users for a campaign (last 7d) ─────── +-- Used by GET /api/auth-events/count to verify campaign conversion. +create or replace function public.wavex_os_count_campaign_signups( + p_utm_campaign text, + p_days int default 7 +) +returns int +language sql +security definer +set search_path = '' +as $$ + select count(distinct user_id)::int + from wavex_os.auth_events + where utm_campaign = p_utm_campaign + and event_type = 'signup_confirmed' + and created_at >= now() - (p_days || ' days')::interval; +$$; + +grant execute on function public.wavex_os_count_campaign_signups(text, int) + to service_role; diff --git a/tools/backfill-resend-audience.mjs b/tools/backfill-resend-audience.mjs new file mode 100755 index 0000000..8f8399b --- /dev/null +++ b/tools/backfill-resend-audience.mjs @@ -0,0 +1,193 @@ +#!/usr/bin/env node +/** + * Backfill Resend audience contacts from historic auth_events. + * + * Finds all wavex_os.auth_events rows where: + * resend_fired = false AND event_type = 'signup_confirmed' AND email IS NOT NULL + * and adds each as a contact in the configured Resend audience, then marks + * resend_fired = true in the DB. + * + * Requires env vars (loaded from ~/.wavex-os/state/.env if not already set): + * RESEND_API_KEY — Resend API key + * RESEND_AUDIENCE_ID — Resend audience UUID + * SUPABASE_URL — project URL + * SUPABASE_SERVICE_ROLE_KEY — service-role key (full DB access) + * + * Dry-run (safe default — no writes): + * node tools/backfill-resend-audience.mjs + * + * Live run: + * DRY_RUN=false node tools/backfill-resend-audience.mjs + */ + +import fs from 'fs'; +import path from 'path'; +import os from 'os'; + +// ─── env loading ───────────────────────────────────────────────────────────── + +function loadEnvFile(filePath) { + if (!fs.existsSync(filePath)) return; + const lines = fs.readFileSync(filePath, 'utf8').split('\n'); + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith('#')) continue; + const idx = trimmed.indexOf('='); + if (idx === -1) continue; + const key = trimmed.slice(0, idx).trim(); + let val = trimmed.slice(idx + 1).trim(); + val = val.replace(/^["']|["']$/g, ''); + if (!process.env[key]) process.env[key] = val; + } +} + +loadEnvFile(path.join(os.homedir(), '.wavex-os', 'state', '.env')); + +// ─── config ─────────────────────────────────────────────────────────────────── + +const DRY_RUN = process.env.DRY_RUN !== 'false'; +const RESEND_API_KEY = process.env.RESEND_API_KEY; +const RESEND_AUDIENCE_ID = process.env.RESEND_AUDIENCE_ID; +const SUPABASE_URL = process.env.SUPABASE_URL; +const SUPABASE_KEY = process.env.SUPABASE_SERVICE_ROLE_KEY; +const BATCH_SIZE = 100; +const DELAY_MS = 300; // polite delay between Resend API calls + +if (!SUPABASE_URL || !SUPABASE_KEY) { + console.error('SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY are required.'); + process.exit(1); +} +if (!DRY_RUN && (!RESEND_API_KEY || !RESEND_AUDIENCE_ID)) { + console.error('RESEND_API_KEY and RESEND_AUDIENCE_ID are required for live runs.'); + console.error('Run with DRY_RUN=false after setting these in ~/.wavex-os/state/.env'); + process.exit(1); +} + +// ─── Supabase helpers ───────────────────────────────────────────────────────── +// wavex_os schema is not exposed via PostgREST — use SECURITY DEFINER RPCs. + +async function supabaseRpc(fn, params) { + const res = await fetch(`${SUPABASE_URL}/rest/v1/rpc/${fn}`, { + method: 'POST', + headers: { + apikey: SUPABASE_KEY, + Authorization: `Bearer ${SUPABASE_KEY}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify(params), + }); + if (!res.ok) { + const detail = await res.text().catch(() => ''); + throw new Error(`RPC ${fn} failed: ${res.status} ${detail}`); + } + return res.json(); +} + +async function fetchUnfiredRows() { + return supabaseRpc('wavex_os_get_unfired_auth_events', { p_limit: BATCH_SIZE }); +} + +async function markFired(id) { + await supabaseRpc('wavex_os_mark_auth_event_fired', { p_id: id }); +} + +// ─── Resend helper ──────────────────────────────────────────────────────────── + +async function addContact({ email, utmCampaign }) { + const ctrl = new AbortController(); + const t = setTimeout(() => ctrl.abort(), 10_000); + try { + const body = { email, unsubscribed: false }; + if (utmCampaign) body.data = { utm_campaign: utmCampaign }; + + const res = await fetch( + `https://api.resend.com/audiences/${encodeURIComponent(RESEND_AUDIENCE_ID)}/contacts`, + { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${RESEND_API_KEY}`, + }, + body: JSON.stringify(body), + signal: ctrl.signal, + }, + ); + clearTimeout(t); + if (!res.ok) { + const detail = await res.text().catch(() => ''); + // 409 = contact already exists — treat as success for idempotency + if (res.status === 409) return { ok: true, alreadyExists: true }; + throw new Error(`Resend ${res.status}: ${detail}`); + } + return { ok: true, alreadyExists: false }; + } catch (e) { + clearTimeout(t); + throw e; + } +} + +// ─── main ───────────────────────────────────────────────────────────────────── + +console.log('\n=== Resend Audience Backfill ==='); +console.log(`Mode: ${DRY_RUN ? 'DRY RUN (no writes)' : '*** LIVE — writing to Resend + DB ***'}`); +if (!DRY_RUN) console.log(`Audience: ${RESEND_AUDIENCE_ID}`); +console.log(''); + +let totalProcessed = 0; +let totalFired = 0; +let totalSkipped = 0; +let totalFailed = 0; +let pageOffset = 0; + +while (true) { + const rows = await fetchUnfiredRows(pageOffset); + if (rows.length === 0) break; + + console.log(`Batch: ${rows.length} rows (total so far: ${totalProcessed})`); + + for (const row of rows) { + totalProcessed++; + const label = `${row.email} (id=${row.id.slice(0, 8)}… campaign=${row.utm_campaign ?? 'none'})`; + + if (DRY_RUN) { + console.log(` [DRY RUN] Would add: ${label}`); + totalFired++; + continue; + } + + try { + const result = await addContact({ email: row.email, utmCampaign: row.utm_campaign }); + if (result.alreadyExists) { + console.log(` ↩ already exists: ${label}`); + totalSkipped++; + } else { + console.log(` ✅ added: ${label}`); + totalFired++; + } + await markFired(row.id); + } catch (e) { + console.error(` ❌ failed: ${label} — ${e.message}`); + totalFailed++; + } + + if (rows.indexOf(row) < rows.length - 1) { + await new Promise(r => setTimeout(r, DELAY_MS)); + } + } + + // If we got a full batch, there may be more rows — but since we mark fired=true, + // re-querying from offset 0 is correct (processed rows no longer match). + if (rows.length < BATCH_SIZE) break; +} + +console.log('\n=== Summary ==='); +console.log(`Processed : ${totalProcessed}`); +if (DRY_RUN) { + console.log(`Would fire: ${totalFired}`); + console.log('\nRe-run with DRY_RUN=false to execute.'); +} else { + console.log(`Fired : ${totalFired}`); + console.log(`Skipped : ${totalSkipped} (already in audience)`); + console.log(`Failed : ${totalFailed}`); + if (totalFailed > 0) process.exit(1); +}