From 9edb18db601a28f0c379dbe859e52749bcbbe84f Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Sat, 8 Aug 2026 19:17:41 -0400 Subject: [PATCH] feat(gdpr): user data erasure workflow (EVE-46) (#468) --- .env.production.example | 2 + alchemy.run.ts | 1 + package.json | 1 + runbooks/gdpr-erasure.md | 101 +++ scripts/erase-user-data.ts | 664 ++++++++++++++++++ src/env.d.ts | 2 + src/server.ts | 6 + .../ai-search/services/promptExplorer.ts | 13 +- src/server/features/audit/AuditScratchpad.ts | 5 + .../onboarding/OnboardingChatAgent.ts | 12 + src/server/features/sam/SamChatAgent.ts | 11 + src/server/gdpr/storage-erasure.ts | 296 ++++++++ src/server/lib/r2-cache.ts | 9 + src/shared/gdpr-erasure.test.ts | 18 + src/shared/gdpr-erasure.ts | 62 ++ 15 files changed, 1200 insertions(+), 3 deletions(-) create mode 100644 runbooks/gdpr-erasure.md create mode 100644 scripts/erase-user-data.ts create mode 100644 src/server/gdpr/storage-erasure.ts create mode 100644 src/shared/gdpr-erasure.test.ts create mode 100644 src/shared/gdpr-erasure.ts diff --git a/.env.production.example b/.env.production.example index a50b75e..7460a8a 100644 --- a/.env.production.example +++ b/.env.production.example @@ -32,6 +32,8 @@ DATABASE_PROVIDER=postgres # Billing, analytics, agents. # AUTUMN_SECRET_KEY= # AUTUMN_WEBHOOK_SECRET= +# Long random secret shared only with scripts/erase-user-data.ts. +# GDPR_ERASURE_SECRET= # POSTHOG_PUBLIC_KEY= # POSTHOG_HOST= # OPENROUTER_API_KEY= diff --git a/alchemy.run.ts b/alchemy.run.ts index 170c1f6..a415770 100644 --- a/alchemy.run.ts +++ b/alchemy.run.ts @@ -271,6 +271,7 @@ const dataEnv = { OPENROUTER_MODEL: optionalVar("OPENROUTER_MODEL"), AUTUMN_SECRET_KEY: optionalSecret("AUTUMN_SECRET_KEY"), AUTUMN_WEBHOOK_SECRET: optionalSecret("AUTUMN_WEBHOOK_SECRET"), + GDPR_ERASURE_SECRET: optionalSecret("GDPR_ERASURE_SECRET"), LOOPS_API_KEY: optionalSecret("LOOPS_API_KEY"), LOOPS_TRANSACTIONAL_VERIFY_EMAIL_ID: optionalVar( "LOOPS_TRANSACTIONAL_VERIFY_EMAIL_ID", diff --git a/package.json b/package.json index 0d4799b..259e273 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "billing:brand-lookup": "tsx scripts/brand-lookup-cost-profile.ts", "billing:usage": "tsx scripts/dataforseo-account-usage.ts", "cleanup:default-projects:d1": "tsx scripts/d1-default-project-cleanup.ts", + "gdpr:erase-user": "tsx scripts/erase-user-data.ts", "seed:rank-tracking": "tsx scripts/seed-rank-tracking.ts", "seed:projects": "tsx scripts/seed-projects.ts", "ci:check": "prettier --check . && knip && tsc --noEmit && tsc --noEmit -p badseo/tsconfig.json && oxlint . --type-aware" diff --git a/runbooks/gdpr-erasure.md b/runbooks/gdpr-erasure.md new file mode 100644 index 0000000..d55c238 --- /dev/null +++ b/runbooks/gdpr-erasure.md @@ -0,0 +1,101 @@ +# GDPR user erasure + +`pnpm gdpr:erase-user` inventories and erases a hosted user's OpenSEO data. It +is Postgres-only and covers the application database, Cloudflare-bound state, +Google grants, Loops, PostHog, Autumn, and the Stripe customer linked through +Autumn. + +The command defaults to a dry run. It refuses any organization with more than +one current member because deleting that workspace would erase another user's +data. Resolve ownership and shared-data retention manually before retrying. + +## One-time deployment setup + +Generate a long random secret and set the same value in the production Worker +and in the operator environment as `GDPR_ERASURE_SECRET`. The secret enables a +single HMAC-authenticated endpoint at `/api/internal/gdpr-erasure/storage`; the +endpoint returns 404 when the secret is unset. + +Add `GDPR_ERASURE_SECRET` to `.env.production`, then deploy the Worker before +using the command. Keep the secret out of shell history and logs. + +Verify the production R2 bucket has a lifecycle rule expiring the +`dataforseo-cache/` prefix (prod lifecycle rules are dashboard-managed). The +retention claims below depend on it. + +## Operator environment + +Put these values in a secure, untracked environment file or secret manager: + +```dotenv +POSTGRES_DATABASE_URL=postgres://... +BETTER_AUTH_URL=https://app.openseo.so +GDPR_ERASURE_SECRET=... + +LOOPS_API_KEY=... +AUTUMN_SECRET_KEY=... + +POSTHOG_API_HOST=https://us.posthog.com +POSTHOG_PROJECT_ID=... +POSTHOG_PERSONAL_API_KEY=... +``` + +The script also reads `.env.local`/`.env` from the working directory, so run it +from a directory whose env files hold the intended values. The PostHog personal +key needs person read and write access. Use the EU PostHog API host if that is +where the project lives. + +## Run + +First inventory the exact target and affected row/resource counts: + +```bash +pnpm gdpr:erase-user --email person@example.com +``` + +Confirm the identity and inventory, and that the dry run prints +`autumnEnvironment: "live"` — a sandbox Autumn key would silently skip the +production Stripe deletion. Then execute with the exact normalized email and +database host printed by the dry run: + +```bash +pnpm gdpr:erase-user --email person@example.com \ + --execute --confirm person@example.com \ + --confirm-database-host us-east-3.pg.psdb.cloud +``` + +`--user-id ` can replace `--email` as the selector, but `--confirm` still +must be the email printed in the inventory. + +The execution order is designed for safe retries: + +1. Delete the Loops contact, queue PostHog person/event deletion, and delete + Autumn plus its linked Stripe customer. +2. Call the Worker endpoint, which terminates active site-audit and rank-check + Workflow instances, revokes Google grants, and erases chat/scratchpad + Durable Objects, R2 audit payloads, KV progress entries, and MCP OAuth + grants/tokens. +3. Delete the organizations and user in one Postgres transaction, relying on + foreign-key cascades for project data, then verify the root rows are gone. + +If a step fails, fix the reported credential or service error and run the same +command again. Vendor absence and already-finished Workflows are treated as +successful no-ops. + +## Retention notes + +The completion JSON is the erasure receipt; save it in the request case without +adding it back to product analytics. PostHog event deletion runs asynchronously. +Completed Workflow state and Workers logs expire under the Cloudflare account's +configured retention. Database backups and billing records that must be kept +for tax, fraud, or legal obligations should be isolated from production access +and allowed to expire under the documented retention schedule. + +Prompt-response cache objects written after this erasure tooling was deployed +carry an organization tag and are deleted by the command. Older untagged cache +objects cannot be attributed to a user from their hashed key; they remain +inaccessible after account deletion and expire under the bucket's +`dataforseo-cache/` lifecycle rule (see the deployment setup above). Audit +scratchpad state for audits older than 30 days is skipped by the command +because those Durable Objects and progress keys already self-destructed via +their finalize path, 7-day alarm, or 30-minute TTL. diff --git a/scripts/erase-user-data.ts b/scripts/erase-user-data.ts new file mode 100644 index 0000000..5781a9a --- /dev/null +++ b/scripts/erase-user-data.ts @@ -0,0 +1,664 @@ +/** + * GDPR account erasure for the hosted Postgres deployment. + * + * Dry run (the default): + * pnpm gdpr:erase-user --email person@example.com + * + * Execute after reviewing the inventory: + * pnpm gdpr:erase-user --email person@example.com \ + * --execute --confirm person@example.com \ + * --confirm-database-host + * + * The Worker endpoint must be deployed with the same GDPR_ERASURE_SECRET as + * this process. See runbooks/gdpr-erasure.md for required operator variables. + */ +import process from "node:process"; +import { Autumn } from "autumn-js"; +import { + and, + count, + eq, + gt, + inArray, + isNotNull, + ne, + notExists, + or, + sql, +} from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; +import { drizzle } from "drizzle-orm/postgres-js"; +import postgres from "postgres"; +import { z } from "zod"; +import { GA4_OAUTH_PROVIDER_ID } from "../src/shared/ga4"; +import { + GDPR_STORAGE_ERASURE_PATH, + signGdprErasureRequest, + type GdprStorageErasurePayload, +} from "../src/shared/gdpr-erasure"; +import { GSC_OAUTH_PROVIDER_ID } from "../src/shared/gsc"; +import { loadLocalEnv, parseArgs } from "./cli-utils"; +// The Node-safe raw barrel (not ../src/db/schema, the provider-aware one, +// which imports cloudflare:workers). +import * as schema from "../src/db/pg/schema"; + +loadLocalEnv(); + +const args = parseArgs(process.argv.slice(2)); +const execute = args.execute === "true"; +const emailSelector = args.email?.trim().toLowerCase(); +const userIdSelector = args["user-id"]?.trim(); + +type Db = ReturnType; +type UserRow = { id: string; email: string; name: string }; + +function requiredEnv(name: string): string { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`${name} is required.`); + return value; +} + +function optionalEnv(name: string, fallback: string): string { + return process.env[name]?.trim() || fallback; +} + +function errorText(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function isNotFound(error: unknown): boolean { + const message = errorText(error).toLowerCase(); + return message.includes("404") || message.includes("not found"); +} + +function printUsage(): never { + throw new Error( + "Pass exactly one selector: --email person@example.com or --user-id . Add --execute --confirm only after reviewing the dry run.", + ); +} + +async function findUser(db: Db): Promise { + if (Boolean(emailSelector) === Boolean(userIdSelector)) printUsage(); + const rows = await db + .select({ + id: schema.user.id, + email: schema.user.email, + name: schema.user.name, + }) + .from(schema.user) + .where( + emailSelector + ? sql`lower(${schema.user.email}) = ${emailSelector}` + : eq(schema.user.id, userIdSelector ?? ""), + ) + .limit(2); + if (rows.length === 0) throw new Error("No matching user found."); + if (rows.length !== 1) throw new Error("Selector matched multiple users."); + return rows[0]; +} + +async function buildInventory(db: Db, user: UserRow) { + const allMembers = alias(schema.member, "all_members"); + const organizations = await db + .select({ + id: schema.organization.id, + name: schema.organization.name, + memberCount: count(allMembers.id), + }) + .from(schema.member) + .innerJoin( + schema.organization, + eq(schema.organization.id, schema.member.organizationId), + ) + .innerJoin( + allMembers, + eq(allMembers.organizationId, schema.organization.id), + ) + .where(eq(schema.member.userId, user.id)) + .groupBy(schema.organization.id, schema.organization.name) + .orderBy(schema.organization.id); + const shared = organizations.filter( + (organization) => organization.memberCount !== 1, + ); + if (shared.length > 0) { + throw new Error( + `Refusing to erase shared organization(s): ${shared + .map( + (organization) => + `${organization.id} (${organization.memberCount} members)`, + ) + .join( + ", ", + )}. Transfer/remove the user and handle shared records explicitly first.`, + ); + } + const organizationIds = organizations.map((organization) => organization.id); + + // drizzle's inArray throws on empty arrays, so project-scoped queries are + // skipped outright when the user has no organizations or projects. + const projects = + organizationIds.length === 0 + ? [] + : await db + .select({ id: schema.projects.id }) + .from(schema.projects) + .where(inArray(schema.projects.organizationId, organizationIds)) + .orderBy(schema.projects.id); + const projectIds = projects.map((row) => row.id); + + const samSessions = await db + .select({ id: schema.samSessions.id }) + .from(schema.samSessions) + .where(eq(schema.samSessions.userId, user.id)) + .orderBy(schema.samSessions.id); + + // Scratchpad DOs self-destroy at finalize and via a 7-day alarm, and the + // audit-progress KV key has a 30-minute TTL, so older audits have no + // Cloudflare state left to erase. 30 days gives 4x margin over the alarm + // and keeps the Worker call within KV's per-invocation operation limit. + // startedAt is a text column of ISO strings, so compare lexicographically. + const auditCutoff = new Date( + Date.now() - 30 * 24 * 60 * 60 * 1000, + ).toISOString(); + const audits = + projectIds.length === 0 + ? [] + : await db + .select({ id: schema.audits.id }) + .from(schema.audits) + .where( + and( + inArray(schema.audits.projectId, projectIds), + or( + gt(schema.audits.startedAt, auditCutoff), + eq(schema.audits.status, "running"), + ), + ), + ) + .orderBy(schema.audits.id); + + const r2Rows = + projectIds.length === 0 + ? [] + : await db + .selectDistinct({ r2Key: schema.auditLighthouseResults.r2Key }) + .from(schema.auditLighthouseResults) + .innerJoin( + schema.audits, + eq(schema.audits.id, schema.auditLighthouseResults.auditId), + ) + .where( + and( + inArray(schema.audits.projectId, projectIds), + isNotNull(schema.auditLighthouseResults.r2Key), + ), + ) + .orderBy(schema.auditLighthouseResults.r2Key); + const r2Keys = r2Rows.flatMap((row) => (row.r2Key ? [row.r2Key] : [])); + + const googleAccountRows = await db + .selectDistinct({ + providerId: schema.account.providerId, + accountId: schema.account.accountId, + }) + .from(schema.account) + .where( + and( + eq(schema.account.userId, user.id), + inArray(schema.account.providerId, [ + GSC_OAUTH_PROVIDER_ID, + GA4_OAUTH_PROVIDER_ID, + ]), + ), + ) + .orderBy(schema.account.providerId, schema.account.accountId); + // The where clause already restricts providerId to the two Google + // providers; the predicate narrows the column's string type to match the + // erasure payload's enum. + const isGoogleProviderId = ( + value: string, + ): value is typeof GSC_OAUTH_PROVIDER_ID | typeof GA4_OAUTH_PROVIDER_ID => + value === GSC_OAUTH_PROVIDER_ID || value === GA4_OAUTH_PROVIDER_ID; + const googleAccounts = googleAccountRows.flatMap((row) => + isGoogleProviderId(row.providerId) + ? [{ providerId: row.providerId, accountId: row.accountId }] + : [], + ); + + const activeAuditWorkflows = + projectIds.length === 0 + ? [] + : await db + .select({ + id: sql`coalesce(${schema.audits.workflowInstanceId}, ${schema.audits.id})`, + }) + .from(schema.audits) + .where( + and( + inArray(schema.audits.projectId, projectIds), + eq(schema.audits.status, "running"), + ), + ) + .orderBy(schema.audits.id); + const activeRankWorkflows = + projectIds.length === 0 + ? [] + : await db + .select({ id: schema.rankCheckRuns.id }) + .from(schema.rankCheckRuns) + .where( + and( + inArray(schema.rankCheckRuns.projectId, projectIds), + inArray(schema.rankCheckRuns.status, ["pending", "running"]), + ), + ) + .orderBy(schema.rankCheckRuns.id); + + const projectCount = async (table: typeof schema.savedKeywords) => + projectIds.length === 0 + ? 0 + : db.$count(table, inArray(table.projectId, projectIds)); + const databaseCounts = { + sessions: await db.$count( + schema.session, + eq(schema.session.userId, user.id), + ), + accounts: await db.$count( + schema.account, + eq(schema.account.userId, user.id), + ), + onboarding_answers: await db.$count( + schema.userOnboardingAnswers, + eq(schema.userOnboardingAnswers.userId, user.id), + ), + projects: projectIds.length, + saved_keywords: await projectCount(schema.savedKeywords), + audits: + projectIds.length === 0 + ? 0 + : await db.$count( + schema.audits, + inArray(schema.audits.projectId, projectIds), + ), + rank_snapshots: + projectIds.length === 0 + ? 0 + : await db + .select({ value: count() }) + .from(schema.rankSnapshots) + .innerJoin( + schema.rankCheckRuns, + eq(schema.rankCheckRuns.id, schema.rankSnapshots.runId), + ) + .where(inArray(schema.rankCheckRuns.projectId, projectIds)) + .then((rows) => rows[0]?.value ?? 0), + sam_sessions: samSessions.length, + attributed_audits: await db.$count( + schema.audits, + eq(schema.audits.startedByUserId, user.id), + ), + gsc_connections: await db.$count( + schema.gscConnections, + eq(schema.gscConnections.connectedByUserId, user.id), + ), + ga4_connections: await db.$count( + schema.ga4Connections, + eq(schema.ga4Connections.connectedByUserId, user.id), + ), + reddit_attributions: await db.$count( + schema.redditAttributions, + eq(schema.redditAttributions.userId, user.id), + ), + }; + + return { + organizations, + projectIds, + samSessionIds: samSessions.map((row) => row.id), + auditIds: audits.map((row) => row.id), + r2Keys, + googleAccounts, + activeAuditWorkflowIds: activeAuditWorkflows.map((row) => row.id), + activeRankWorkflowIds: activeRankWorkflows.map((row) => row.id), + databaseCounts, + }; +} + +async function deleteLoopsContactBy(selector: { + userId?: string; + email?: string; +}) { + const response = await fetch("https://app.loops.so/api/v1/contacts/delete", { + method: "POST", + headers: { + Authorization: `Bearer ${requiredEnv("LOOPS_API_KEY")}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(selector), + }); + if (response.status === 404) return "already_absent"; + if (!response.ok) { + throw new Error( + `Loops deletion failed (${response.status}): ${await response.text()}`, + ); + } + return "deleted"; +} + +async function deleteLoopsContact(userId: string, email: string) { + // Waitlist signups (web/src/routes/api/subscribe.ts) create contacts with + // email only, and the app-side userId backfill is best-effort — so delete by + // each selector. The Loops API rejects a request carrying both. + return { + byUserId: await deleteLoopsContactBy({ userId }), + byEmail: await deleteLoopsContactBy({ email }), + }; +} + +async function deletePostHogPerson(userId: string) { + const host = optionalEnv( + "POSTHOG_API_HOST", + "https://us.posthog.com", + ).replace(/\/$/u, ""); + const projectId = encodeURIComponent(requiredEnv("POSTHOG_PROJECT_ID")); + const authorization = `Bearer ${requiredEnv("POSTHOG_PERSONAL_API_KEY")}`; + // A distinct_id filter matches at most one person, so a single page is + // enough; `next` is deliberately ignored. + const listUrl = new URL(`${host}/api/projects/${projectId}/persons/`); + listUrl.searchParams.set("distinct_id", userId); + const listResponse = await fetch(listUrl, { + headers: { Authorization: authorization }, + }); + if (!listResponse.ok) { + throw new Error( + `PostHog lookup failed (${listResponse.status}): ${await listResponse.text()}`, + ); + } + const people = z + .object({ + results: z.array( + z.object({ + id: z.union([z.string(), z.number()]).optional(), + uuid: z.string().optional(), + }), + ), + }) + .parse(await listResponse.json()).results; + for (const person of people) { + const personId = person.id ?? person.uuid; + if (personId === undefined) + throw new Error("PostHog returned a person without an id."); + const response = await fetch( + `${host}/api/projects/${projectId}/persons/${encodeURIComponent(String(personId))}/?delete_events=true`, + { method: "DELETE", headers: { Authorization: authorization } }, + ); + if (!response.ok && response.status !== 404) { + throw new Error( + `PostHog deletion failed (${response.status}): ${await response.text()}`, + ); + } + } + return people.length; +} + +/** + * The Autumn key prefix encodes which environment it targets. A sandbox key + * against production data would 404 every delete and leave the live Stripe + * subscription running — surface the environment in the dry run so the + * operator can catch that before executing. + */ +function autumnEnvironment(): string { + const key = process.env.AUTUMN_SECRET_KEY?.trim() ?? ""; + if (key.startsWith("am_sk_live_")) return "live"; + if (key.startsWith("am_sk_test_")) return "sandbox"; + return "unset-or-unknown"; +} + +async function deleteAutumnCustomer(organizationIds: string[]) { + const autumn = new Autumn({ secretKey: requiredEnv("AUTUMN_SECRET_KEY") }); + let deleted = 0; + let absent = 0; + for (const organizationId of organizationIds) { + try { + await autumn.customers.delete({ + customerId: organizationId, + deleteInStripe: true, + }); + deleted += 1; + } catch (error) { + if (!isNotFound(error)) throw error; + absent += 1; + } + } + return { deleted, absent }; +} + +async function eraseWorkerStorage(payload: GdprStorageErasurePayload) { + const secret = requiredEnv("GDPR_ERASURE_SECRET"); + const endpoint = new URL( + GDPR_STORAGE_ERASURE_PATH, + requiredEnv("BETTER_AUTH_URL"), + ); + if (endpoint.protocol !== "https:") { + throw new Error("BETTER_AUTH_URL must use https for GDPR erasure."); + } + const body = JSON.stringify(payload); + const timestamp = String(Date.now()); + const signature = await signGdprErasureRequest(secret, timestamp, body); + const response = await fetch(endpoint, { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-gdpr-timestamp": timestamp, + "x-gdpr-signature": signature, + }, + body, + }); + const responseBody = await response.text(); + if (!response.ok) { + throw new Error( + `Worker storage erasure failed (${response.status}): ${responseBody}`, + ); + } + return z + .object({ ok: z.literal(true), result: z.record(z.string(), z.unknown()) }) + .parse(JSON.parse(responseBody) as unknown).result; +} + +async function erasePostgres(db: Db, user: UserRow, organizationIds: string[]) { + await db.transaction(async (tx) => { + await tx + .delete(schema.invitation) + .where(sql`lower(${schema.invitation.email}) = lower(${user.email})`); + // Better Auth stores verification rows keyed by composite identifiers + // (reset-password:, delete-account-) whose value is the + // user id; emails never appear bare in either column. + await tx + .delete(schema.verification) + .where( + or( + eq(schema.verification.identifier, user.id), + eq(schema.verification.value, user.id), + ), + ); + // These columns intentionally have no user FK. Remove connection records + // and anonymize retained audit attribution even if the user left that + // workspace before making this request. + await tx + .delete(schema.gscConnections) + .where(eq(schema.gscConnections.connectedByUserId, user.id)); + await tx + .delete(schema.ga4Connections) + .where(eq(schema.ga4Connections.connectedByUserId, user.id)); + await tx + .update(schema.audits) + .set({ startedByUserId: "gdpr-deleted-user" }) + .where(eq(schema.audits.startedByUserId, user.id)); + if (organizationIds.length > 0) { + // Re-assert the solo-membership guard at delete time: anyone who + // accepted an invite after the inventory was taken must abort the + // transaction, not be cascaded away with the organization. + const deletedOrganizations = await tx + .delete(schema.organization) + .where( + and( + inArray(schema.organization.id, organizationIds), + notExists( + tx + .select({ one: sql`1` }) + .from(schema.member) + .where( + and( + eq(schema.member.organizationId, schema.organization.id), + ne(schema.member.userId, user.id), + ), + ), + ), + ), + ) + .returning({ id: schema.organization.id }); + if (deletedOrganizations.length !== organizationIds.length) { + throw new Error( + "Organization(s) gained other members since the inventory was taken; aborting the Postgres delete. Re-run after resolving membership.", + ); + } + } + const deleted = await tx + .delete(schema.user) + .where(eq(schema.user.id, user.id)) + .returning({ id: schema.user.id }); + if (deleted.length !== 1) { + throw new Error("Postgres user row disappeared before the final delete."); + } + }); +} + +/** Independent post-commit read-back for the erasure receipt. */ +async function verifyPostgres( + db: Db, + userId: string, + organizationIds: string[], +) { + const userRows = await db.$count(schema.user, eq(schema.user.id, userId)); + const organizationRows = + organizationIds.length === 0 + ? 0 + : await db.$count( + schema.organization, + inArray(schema.organization.id, organizationIds), + ); + if (userRows !== 0 || organizationRows !== 0) { + throw new Error( + "Postgres verification failed: user or organization rows remain.", + ); + } + return { userRows, organizationRows }; +} + +async function main() { + const connectionString = requiredEnv("POSTGRES_DATABASE_URL"); + const databaseUrl = new URL(connectionString); + if (!["postgres:", "postgresql:"].includes(databaseUrl.protocol)) { + throw new Error( + "POSTGRES_DATABASE_URL must use postgres:// or postgresql://.", + ); + } + const databaseHost = databaseUrl.hostname; + const client = postgres(connectionString, { max: 1 }); + const db = drizzle(client); + try { + const user = await findUser(db); + const inventory = await buildInventory(db, user); + const summary = { + mode: execute ? "execute" : "dry-run", + databaseHost, + user: { id: user.id, email: user.email, name: user.name }, + organizations: inventory.organizations, + databaseCounts: inventory.databaseCounts, + cloudflare: { + onboardingChats: inventory.projectIds.length, + samChats: inventory.samSessionIds.length, + auditScratchpads: inventory.auditIds.length, + r2Objects: inventory.r2Keys.length, + activeAuditWorkflows: inventory.activeAuditWorkflowIds.length, + activeRankWorkflows: inventory.activeRankWorkflowIds.length, + }, + external: { + googleAccounts: inventory.googleAccounts.length, + loopsContact: true, + postHogDistinctId: user.id, + autumnCustomersAndStripeCustomers: inventory.organizations.length, + autumnEnvironment: autumnEnvironment(), + }, + }; + console.log(JSON.stringify(summary, null, 2)); + if (!execute) { + console.log( + "\nDry run only. Re-run with --execute --confirm after review.", + ); + return; + } + if (args.confirm?.trim().toLowerCase() !== user.email.toLowerCase()) { + throw new Error( + "--confirm must exactly match the selected user's email.", + ); + } + if (args["confirm-database-host"]?.trim() !== databaseHost) { + throw new Error( + `--confirm-database-host must exactly match ${databaseHost}.`, + ); + } + + const organizationIds = inventory.organizations.map( + (organization) => organization.id, + ); + const loops = await deleteLoopsContact(user.id, user.email); + const postHogPeopleQueued = await deletePostHogPerson(user.id); + const autumnResult = await deleteAutumnCustomer(organizationIds); + const storage = await eraseWorkerStorage({ + userId: user.id, + email: user.email, + organizationIds, + projectIds: inventory.projectIds, + samSessionIds: inventory.samSessionIds, + auditIds: inventory.auditIds, + activeAuditWorkflowIds: inventory.activeAuditWorkflowIds, + activeRankWorkflowIds: inventory.activeRankWorkflowIds, + r2Keys: inventory.r2Keys, + googleAccounts: inventory.googleAccounts, + }); + await erasePostgres(db, user, organizationIds); + const postgresVerification = await verifyPostgres( + db, + user.id, + organizationIds, + ); + + console.log( + JSON.stringify( + { + completedAt: new Date().toISOString(), + userId: user.id, + email: user.email, + vendors: { loops, postHogPeopleQueued, autumn: autumnResult }, + storage, + postgres: postgresVerification, + retentionNotes: [ + "PostHog event deletion is asynchronous after the person deletion request.", + "Completed Cloudflare Workflow state and Workers logs expire under the account retention policy.", + "Database backups and financial records remain only for their documented legal/backup retention periods.", + ], + }, + null, + 2, + ), + ); + } finally { + await client.end(); + } +} + +main().catch((error: unknown) => { + console.error(`GDPR erasure failed: ${errorText(error)}`); + process.exitCode = 1; +}); diff --git a/src/env.d.ts b/src/env.d.ts index 3402041..71e46e5 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -35,6 +35,8 @@ declare namespace Cloudflare { LOOPS_TRANSACTIONAL_RESET_PASSWORD_ID?: string; AUTUMN_SECRET_KEY?: string; AUTUMN_WEBHOOK_SECRET?: string; + // HMAC secret for the operator-only GDPR storage-erasure endpoint. + GDPR_ERASURE_SECRET?: string; // Cloudflare Turnstile — signup captcha (hosted only). Secret verifies // tokens server-side; site key is public and inlined into the client build. diff --git a/src/server.ts b/src/server.ts index 854e22b..c838647 100644 --- a/src/server.ts +++ b/src/server.ts @@ -24,6 +24,8 @@ import { handleAutumnWebhookRequest, } from "@/server/billing/autumn-webhook"; import { maybeSendSelfHostHeartbeat } from "@/server/lib/self-host-telemetry"; +import { handleGdprStorageErasure } from "@/server/gdpr/storage-erasure"; +import { GDPR_STORAGE_ERASURE_PATH } from "@/shared/gdpr-erasure"; const appFetch = createStartHandler(defaultStreamHandler); const openSeoOAuthProvider = createOpenSeoOAuthProvider(appFetch); @@ -145,6 +147,10 @@ function handleFetch( const publicRequest = requestWithPublicOrigin(request); const pathname = new URL(publicRequest.url).pathname; + if (pathname === GDPR_STORAGE_ERASURE_PATH) { + return handleGdprStorageErasure(publicRequest, env); + } + if (pathname.startsWith("/agents/")) { return routeChatAgents(publicRequest, env); } diff --git a/src/server/features/ai-search/services/promptExplorer.ts b/src/server/features/ai-search/services/promptExplorer.ts index 2151687..512d721 100644 --- a/src/server/features/ai-search/services/promptExplorer.ts +++ b/src/server/features/ai-search/services/promptExplorer.ts @@ -3,7 +3,12 @@ import type { BillingCustomerContext } from "@/server/billing/subscription"; import { createDataforseoClient } from "@/server/lib/dataforseo"; import type { LlmResponseResult } from "@/server/lib/dataforseoLlmSchemas"; import { AppError } from "@/server/lib/errors"; -import { buildCacheKey, getCached, setCached } from "@/server/lib/r2-cache"; +import { + AI_SEARCH_PROMPT_CACHE_NAMESPACE, + buildCacheKey, + getCached, + setCached, +} from "@/server/lib/r2-cache"; import { safeHostname, safeHttpUrl } from "@/server/features/ai-search/safeUrl"; import { promptExplorerModelResultSchema, @@ -88,7 +93,7 @@ type RunModelArgs = { async function runModel( args: RunModelArgs, ): Promise { - const cacheKey = await buildCacheKey("ai-search:prompt-response", { + const cacheKey = await buildCacheKey(AI_SEARCH_PROMPT_CACHE_NAMESPACE, { organizationId: args.billingCustomer.organizationId, projectId: args.input.projectId, model: args.model, @@ -115,7 +120,9 @@ async function runModel( const shaped = shapeSuccess(args.model, rawResponse); waitUntil( - setCached(cacheKey, shaped, PROMPT_RESPONSE_TTL_SECONDS).catch((err) => { + setCached(cacheKey, shaped, PROMPT_RESPONSE_TTL_SECONDS, { + organizationId: args.billingCustomer.organizationId, + }).catch((err) => { console.error("ai-search.prompt-response.cache-write failed:", err); }), ); diff --git a/src/server/features/audit/AuditScratchpad.ts b/src/server/features/audit/AuditScratchpad.ts index 1bf3319..ca8c398 100644 --- a/src/server/features/audit/AuditScratchpad.ts +++ b/src/server/features/audit/AuditScratchpad.ts @@ -312,6 +312,11 @@ export class AuditScratchpad extends DurableObject { await this.ctx.storage.deleteAll(); } + /** Same wipe exposed under the common erasure RPC used by the admin tool. */ + async destroyForErasure(): Promise { + await this.destroy(); + } + /** * Self-cleanup for audits whose workflow died without reaching finalize. * Full destroy(), not just deleteAll(): under our compatibility date, diff --git a/src/server/features/onboarding/OnboardingChatAgent.ts b/src/server/features/onboarding/OnboardingChatAgent.ts index a13092c..3f11da8 100644 --- a/src/server/features/onboarding/OnboardingChatAgent.ts +++ b/src/server/features/onboarding/OnboardingChatAgent.ts @@ -71,6 +71,18 @@ export class OnboardingChatAgent extends AIChatAgent { // Cap stored history; the onboarding chat is short and pre-paywall. maxPersistedMessages = 60; + /** Permanently remove this project's transcript for an account erasure. */ + async destroyForErasure(): Promise { + for (const socket of this.ctx.getWebSockets()) { + socket.close(1000, "Account erased"); + } + this.abortAllRequests("GDPR erasure"); + this.resetTurnState(); + await this.waitUntilStable({ timeout: 5_000 }); + await this.ctx.storage.deleteAlarm(); + await this.ctx.storage.deleteAll(); + } + // The base class persists each message as its own bounded SQLite row, so DO // storage occasionally returns a transient internal error (code 10001) that // clears on retry. Retry the message-write path a couple of times before diff --git a/src/server/features/sam/SamChatAgent.ts b/src/server/features/sam/SamChatAgent.ts index 950a91b..0f99e02 100644 --- a/src/server/features/sam/SamChatAgent.ts +++ b/src/server/features/sam/SamChatAgent.ts @@ -99,6 +99,17 @@ export class SamChatAgent extends Think { private turnCostUsd = 0; private turnMonthlyRemaining: number | null = null; + /** Permanently remove this session's transcript for an account erasure. */ + async destroyForErasure(): Promise { + for (const socket of this.ctx.getWebSockets()) { + socket.close(1000, "Account erased"); + } + this.cancelAllChats(); + await this.waitUntilStable({ timeout: 5000 }); + await this.ctx.storage.deleteAlarm(); + await this.ctx.storage.deleteAll(); + } + // Record the app origin for the deep links tools attach to responses, // derived from the requests this DO serves instead of env config. DO storage // (not an instance field) because the DO hibernates: a turn can arrive as a diff --git a/src/server/gdpr/storage-erasure.ts b/src/server/gdpr/storage-erasure.ts new file mode 100644 index 0000000..19603d9 --- /dev/null +++ b/src/server/gdpr/storage-erasure.ts @@ -0,0 +1,296 @@ +import { getAuth } from "@/lib/auth"; +import { getAuditScratchpad } from "@/server/features/audit/AuditScratchpad"; +import type { OnboardingChatAgent } from "@/server/features/onboarding/OnboardingChatAgent"; +import type { SamChatAgent } from "@/server/features/sam/SamChatAgent"; +import { captureServerError } from "@/server/lib/posthog"; +import { + AI_SEARCH_PROMPT_CACHE_NAMESPACE, + cacheObjectPrefix, +} from "@/server/lib/r2-cache"; +import { + gdprStorageErasurePayloadSchema, + signGdprErasureRequest, + type GdprStorageErasurePayload, +} from "@/shared/gdpr-erasure"; + +const MAX_CLOCK_SKEW_MS = 5 * 60 * 1000; +const MAX_BODY_BYTES = 5 * 1024 * 1024; +const GOOGLE_REVOKE_URL = "https://oauth2.googleapis.com/revoke"; +const PROMPT_CACHE_PREFIX = cacheObjectPrefix(AI_SEARCH_PROMPT_CACHE_NAMESPACE); + +type GoogleRevocationResult = { + providerId: string; + accountId: string; + status: "revoked" | "token_unavailable"; +}; + +function timingSafeEqual(left: string, right: string): boolean { + const leftBytes = new TextEncoder().encode(left); + const rightBytes = new TextEncoder().encode(right); + let difference = leftBytes.length ^ rightBytes.length; + const length = Math.max(leftBytes.length, rightBytes.length); + for (let index = 0; index < length; index += 1) { + difference |= (leftBytes[index] ?? 0) ^ (rightBytes[index] ?? 0); + } + return difference === 0; +} + +async function authenticateRequest( + request: Request, + body: string, + secret: string, +): Promise { + const timestamp = request.headers.get("x-gdpr-timestamp") ?? ""; + const signature = request.headers.get("x-gdpr-signature") ?? ""; + const timestampMs = Number(timestamp); + if ( + !timestamp || + !signature || + !Number.isFinite(timestampMs) || + Math.abs(Date.now() - timestampMs) > MAX_CLOCK_SKEW_MS + ) { + return false; + } + const expected = await signGdprErasureRequest(secret, timestamp, body); + return timingSafeEqual(signature, expected); +} + +async function terminateWorkflows(workflow: Workflow, ids: string[]) { + let terminated = 0; + for (const id of ids) { + // get() throws for an unknown id; terminate() throws once the instance + // reached a terminal state. Both mean there is nothing left to stop. + const instance = await workflow.get(id).catch(() => null); + if (!instance) continue; + try { + await instance.terminate(); + terminated += 1; + } catch { + // Already complete, errored, or terminated. + } + } + return terminated; +} + +async function deleteKvPrefix(namespace: KVNamespace, prefix: string) { + let cursor: string | undefined; + let deleted = 0; + do { + const page = await namespace.list({ prefix, cursor }); + for (const key of page.keys) { + await namespace.delete(key.name); + deleted += 1; + } + cursor = page.list_complete ? undefined : page.cursor; + } while (cursor); + return deleted; +} + +async function deleteOauthGrants(namespace: KVNamespace, userId: string) { + const grantPrefix = `grant:${userId}:`; + let cursor: string | undefined; + let deletedGrants = 0; + let deletedTokens = 0; + do { + const page = await namespace.list({ prefix: grantPrefix, cursor }); + for (const key of page.keys) { + const grantId = key.name.slice(grantPrefix.length); + deletedTokens += await deleteKvPrefix( + namespace, + `token:${userId}:${grantId}:`, + ); + await namespace.delete(key.name); + deletedGrants += 1; + } + cursor = page.list_complete ? undefined : page.cursor; + } while (cursor); + return { deletedGrants, deletedTokens }; +} + +async function deleteOrganizationPromptCaches( + bucket: R2Bucket, + organizationIds: string[], +) { + const targets = new Set(organizationIds); + let cursor: string | undefined; + const keys: string[] = []; + do { + const page = await bucket.list({ + prefix: PROMPT_CACHE_PREFIX, + cursor, + include: ["customMetadata"], + }); + for (const object of page.objects) { + if (targets.has(object.customMetadata?.organizationId ?? "")) { + keys.push(object.key); + } + } + cursor = page.truncated ? page.cursor : undefined; + } while (cursor); + for (let index = 0; index < keys.length; index += 1_000) { + await bucket.delete(keys.slice(index, index + 1_000)); + } + return keys.length; +} + +async function revokeGoogleAccount( + userId: string, + account: GdprStorageErasurePayload["googleAccounts"][number], +): Promise { + let accessToken: string | undefined; + try { + const result = await getAuth().api.getAccessToken({ + body: { + userId, + providerId: account.providerId, + accountId: account.accountId, + }, + }); + accessToken = result?.accessToken; + } catch { + // If Better Auth cannot mint a token, the locally stored grant is no longer + // usable. The Postgres transaction still removes its encrypted token row. + return { ...account, status: "token_unavailable" }; + } + if (!accessToken) return { ...account, status: "token_unavailable" }; + + const response = await fetch(GOOGLE_REVOKE_URL, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ token: accessToken }), + }); + // Google uses invalid_token for an already-revoked token. Either response + // leaves OpenSEO without a live upstream grant once the local row is erased. + if (!response.ok && response.status !== 400) { + throw new Error( + `Google token revocation failed for ${account.providerId}/${account.accountId}: ${response.status}`, + ); + } + return { ...account, status: "revoked" }; +} + +async function eraseStorage(env: Env, payload: GdprStorageErasurePayload) { + // Stop live workflows first so a running crawl can't rewrite a scratchpad + // after it is wiped. + const auditWorkflowsTerminated = await terminateWorkflows( + env.SITE_AUDIT_WORKFLOW, + payload.activeAuditWorkflowIds, + ); + const rankWorkflowsTerminated = await terminateWorkflows( + env.RANK_CHECK_WORKFLOW, + payload.activeRankWorkflowIds, + ); + + const googleRevocations: GoogleRevocationResult[] = []; + for (const account of payload.googleAccounts) { + googleRevocations.push(await revokeGoogleAccount(payload.userId, account)); + } + + // env.d.ts declares the DO bindings untyped (ambient contexts can't import + // the classes); narrow here so the erasure RPCs are typed. + const samChat = + // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- the binding is declared as this class in wrangler.jsonc + env.SAM_CHAT as unknown as DurableObjectNamespace; + for (const sessionId of payload.samSessionIds) { + await samChat.get(samChat.idFromName(sessionId)).destroyForErasure(); + } + const onboardingChat = + // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- the binding is declared as this class in wrangler.jsonc + env.ONBOARDING_CHAT as unknown as DurableObjectNamespace; + for (const projectId of payload.projectIds) { + await onboardingChat + .get(onboardingChat.idFromName(projectId)) + .destroyForErasure(); + } + for (const auditId of payload.auditIds) { + await getAuditScratchpad(auditId).destroyForErasure(); + await env.KV.delete(`audit-progress:${auditId}`); + } + // The autumn:customer-ensured KV markers are deliberately left alone: they + // hold no personal data (org id key, "1" value), expire on their own 24h + // TTL, and clearing them would let an in-flight authenticated request + // re-create the Autumn customer before the Postgres delete lands. + + for (let index = 0; index < payload.r2Keys.length; index += 1_000) { + await env.R2.delete(payload.r2Keys.slice(index, index + 1_000)); + } + const promptCacheObjects = await deleteOrganizationPromptCaches( + env.R2, + payload.organizationIds, + ); + + const oauth = await deleteOauthGrants(env.OAUTH_KV, payload.userId); + return { + workflows: { + auditTerminated: auditWorkflowsTerminated, + rankTerminated: rankWorkflowsTerminated, + }, + durableObjects: { + sam: payload.samSessionIds.length, + onboarding: payload.projectIds.length, + auditScratchpads: payload.auditIds.length, + }, + kv: { + auditProgress: payload.auditIds.length, + oauthGrants: oauth.deletedGrants, + oauthTokens: oauth.deletedTokens, + }, + r2Objects: payload.r2Keys.length, + promptCacheObjects, + googleRevocations, + }; +} + +export async function handleGdprStorageErasure( + request: Request, + env: Env, +): Promise { + const secret = env.GDPR_ERASURE_SECRET?.trim(); + if (!secret) return new Response("Not found", { status: 404 }); + if (request.method !== "POST") { + return new Response("Method not allowed", { status: 405 }); + } + + // The body is buffered before authentication, so bound it first. workerd + // hands the handler at most content-length bytes (the connection is killed + // at the declared length), while a body with no content-length streams + // unbounded — so the header is required, not advisory. + if (!request.headers.has("content-length")) { + return new Response("Content-Length required", { status: 411 }); + } + const declaredLength = Number(request.headers.get("content-length")); + if ( + !Number.isInteger(declaredLength) || + declaredLength < 0 || + declaredLength > MAX_BODY_BYTES + ) { + return new Response("Payload too large", { status: 413 }); + } + + const rawBody = await request.text(); + if (!(await authenticateRequest(request, rawBody, secret))) { + return new Response("Unauthorized", { status: 401 }); + } + + let input: unknown; + try { + input = JSON.parse(rawBody || "null") as unknown; + } catch { + return Response.json({ error: "Invalid JSON" }, { status: 400 }); + } + const parsed = gdprStorageErasurePayloadSchema.safeParse(input); + if (!parsed.success) { + return Response.json({ error: "Invalid erasure payload" }, { status: 400 }); + } + + try { + const result = await eraseStorage(env, parsed.data); + return Response.json({ ok: true, result }); + } catch (error) { + // Raw fetch handlers run outside the server-function middleware, so + // nothing else reports failures here. + console.error("gdpr.storage-erasure failed:", error); + await captureServerError(error, { source: "gdpr_storage_erasure" }); + return Response.json({ error: "Erasure failed" }, { status: 500 }); + } +} diff --git a/src/server/lib/r2-cache.ts b/src/server/lib/r2-cache.ts index 57a8ab9..fe313ba 100644 --- a/src/server/lib/r2-cache.ts +++ b/src/server/lib/r2-cache.ts @@ -11,6 +11,13 @@ export const CACHE_TTL = { const CACHE_PREFIX = "dataforseo-cache/"; +export const AI_SEARCH_PROMPT_CACHE_NAMESPACE = "ai-search:prompt-response"; + +/** Full R2 object-key prefix for a cache namespace, for prefix listing. */ +export function cacheObjectPrefix(namespace: string): string { + return `${CACHE_PREFIX}${namespace}:`; +} + /** * Build a deterministic cache key from an endpoint slug and input params. * Uses a SHA-256 digest for stability across runtimes. @@ -52,10 +59,12 @@ export async function setCached( key: string, data: T, ttlSeconds: number, + metadata: Record = {}, ): Promise { await env.R2.put(`${CACHE_PREFIX}${key}`, JSON.stringify(data), { httpMetadata: { contentType: "application/json" }, customMetadata: { + ...metadata, expiresAt: new Date(Date.now() + ttlSeconds * 1000).toISOString(), }, }); diff --git a/src/shared/gdpr-erasure.test.ts b/src/shared/gdpr-erasure.test.ts new file mode 100644 index 0000000..b3a963c --- /dev/null +++ b/src/shared/gdpr-erasure.test.ts @@ -0,0 +1,18 @@ +import { createHmac } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { signGdprErasureRequest } from "./gdpr-erasure"; + +describe("GDPR erasure request", () => { + it("signs the timestamp and exact body with HMAC SHA-256", async () => { + const secret = "test-secret"; + const timestamp = "1770000000000"; + const body = '{"userId":"user_1"}'; + const expected = createHmac("sha256", secret) + .update(`${timestamp}.${body}`) + .digest("hex"); + + await expect(signGdprErasureRequest(secret, timestamp, body)).resolves.toBe( + expected, + ); + }); +}); diff --git a/src/shared/gdpr-erasure.ts b/src/shared/gdpr-erasure.ts new file mode 100644 index 0000000..6c5a7c0 --- /dev/null +++ b/src/shared/gdpr-erasure.ts @@ -0,0 +1,62 @@ +import { z } from "zod"; +import { GA4_OAUTH_PROVIDER_ID } from "./ga4"; +import { GSC_OAUTH_PROVIDER_ID } from "./gsc"; + +export const GDPR_STORAGE_ERASURE_PATH = "/api/internal/gdpr-erasure/storage"; + +const MAX_ITEMS_PER_KIND = 10_000; +const boundedIds = z.array(z.string().min(1).max(512)).max(MAX_ITEMS_PER_KIND); + +export const gdprStorageErasurePayloadSchema = z + .object({ + userId: z.string().min(1).max(512), + email: z.string().email().max(512), + organizationIds: boundedIds, + projectIds: boundedIds, + samSessionIds: boundedIds, + auditIds: boundedIds, + activeAuditWorkflowIds: boundedIds, + activeRankWorkflowIds: boundedIds, + r2Keys: boundedIds, + googleAccounts: z + .array( + z.object({ + providerId: z.enum([GSC_OAUTH_PROVIDER_ID, GA4_OAUTH_PROVIDER_ID]), + accountId: z.string().min(1).max(512), + }), + ) + .max(MAX_ITEMS_PER_KIND), + }) + .strict(); + +export type GdprStorageErasurePayload = z.infer< + typeof gdprStorageErasurePayloadSchema +>; + +function toHex(bytes: ArrayBuffer): string { + return [...new Uint8Array(bytes)] + .map((byte) => byte.toString(16).padStart(2, "0")) + .join(""); +} + +export async function signGdprErasureRequest( + secret: string, + timestamp: string, + body: string, +): Promise { + const encoder = new TextEncoder(); + const key = await crypto.subtle.importKey( + "raw", + encoder.encode(secret), + { name: "HMAC", hash: "SHA-256" }, + false, + ["sign"], + ); + return toHex( + await crypto.subtle.sign( + "HMAC", + key, + encoder.encode(`${timestamp}.${body}`), + ), + ); +}