Something went wrong. Try again.
a tool for shared writing and social publishing
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251import { drizzle } from "drizzle-orm/node-postgres";import { eq, inArray, sql } from "drizzle-orm";import { custom_domains, email_auth_tokens, identities, permission_token_on_homepage, publication_email_subscribers, user_entitlements, user_subscriptions,} from "drizzle/schema";import { pool } from "supabase/pool";import { Err, Ok, type Result } from "src/result";import { backfillAtprotoSubscriptionsForIdentity } from "actions/publications/subscribeToPublication";
export type MergeError = | "merge_not_pending" | "invalid_source" | "invalid_target" | "same_identity" | "database_error";
// Merges an email-only identity (source) into an atp-linked identity (target).// Re-verifies invariants under a row lock so concurrent mutations can't// invalidate them mid-transaction.//// NOT a server action — kept in a plain module so it can't be invoked over// the wire. Callers must validate that the requester is authorized to merge// these specific identities before calling.export async function mergeEmailIdentityIntoAtpIdentity(args: { sourceId: string; targetId: string;}): Promise<Result<null, MergeError>> { const { sourceId, targetId } = args; if (sourceId === targetId) return Err("same_identity");
const client = await pool.connect(); let targetAtpDid: string; try { const db = drizzle(client); await db.transaction(async (tx) => { const locked = await tx .select({ id: identities.id, email: identities.email, atp_did: identities.atp_did, }) .from(identities) .where(inArray(identities.id, [sourceId, targetId])) .for("update"); const lockedSource = locked.find((r) => r.id === sourceId); const lockedTarget = locked.find((r) => r.id === targetId); if (!lockedSource || !lockedTarget) throw new Error("merge: identity disappeared under lock"); if (lockedSource.atp_did !== null) throw new Error("merge: source has atp_did"); if (lockedTarget.atp_did === null) throw new Error("merge: target missing atp_did"); if (!lockedSource.email) throw new Error("merge: source missing email"); targetAtpDid = lockedTarget.atp_did;
const sourceEmail = lockedSource.email;
// email_auth_tokens: source's tokens are stale once source is gone. // Caller is responsible for swapping any cookie-held auth_token to a // target-pointing token before calling. await tx .delete(email_auth_tokens) .where(eq(email_auth_tokens.identity, sourceId));
// Target wins on (identity_id) PK collision. await tx.execute(sql` delete from user_subscriptions where identity_id = ${sourceId} and exists (select 1 from user_subscriptions where identity_id = ${targetId}) `); await tx .update(user_subscriptions) .set({ identity_id: targetId }) .where(eq(user_subscriptions.identity_id, sourceId));
// Target wins on (identity_id, entitlement_key) PK collision. await tx.execute(sql` delete from user_entitlements where identity_id = ${sourceId} and entitlement_key in ( select entitlement_key from user_entitlements where identity_id = ${targetId} ) `); await tx .update(user_entitlements) .set({ identity_id: targetId }) .where(eq(user_entitlements.identity_id, sourceId));
// Target wins on (token, identity) PK collision. await tx.execute(sql` delete from permission_token_on_homepage where identity = ${sourceId} and token in ( select token from permission_token_on_homepage where identity = ${targetId} ) `); await tx .update(permission_token_on_homepage) .set({ identity: targetId }) .where(eq(permission_token_on_homepage.identity, sourceId));
// Target wins on unique (publication, email) collision. await tx.execute(sql` delete from ${publication_email_subscribers} where ${publication_email_subscribers.identity_id} = ${sourceId} and (${publication_email_subscribers.publication}, ${publication_email_subscribers.email}) in ( select ${publication_email_subscribers.publication}, ${publication_email_subscribers.email} from ${publication_email_subscribers} where ${publication_email_subscribers.identity_id} = ${targetId} ) `); await tx .update(publication_email_subscribers) .set({ identity_id: targetId }) .where(eq(publication_email_subscribers.identity_id, sourceId));
await tx .update(custom_domains) .set({ identity_id: targetId }) .where(eq(custom_domains.identity_id, sourceId));
// Paid-membership state hangs off identities with ON DELETE CASCADE, so // it must move before the source row is deleted — Stripe keeps billing // the subscriptions either way.
// publication_memberships: unique (publication, identity_id). A live // subscription must keep its row, so on collision the active side wins, // target preferred on a tie. A dropped live row is logged for manual // cancellation; the connect-events webhook also flags it on the // subscription's next event. await tx.execute(sql` delete from publication_memberships t where t.identity_id = ${targetId} and t.status not in ('active', 'trialing') and exists ( select 1 from publication_memberships s where s.identity_id = ${sourceId} and s.publication = t.publication and s.status in ('active', 'trialing') ) `); const droppedMemberships = await tx.execute(sql` delete from publication_memberships s where s.identity_id = ${sourceId} and exists ( select 1 from publication_memberships t where t.identity_id = ${targetId} and t.publication = s.publication ) returning s.publication, s.stripe_subscription_id, s.status `); for (const row of droppedMemberships.rows as { publication: string; stripe_subscription_id: string | null; status: string | null; }[]) { if (row.status === "active" || row.status === "trialing") console.error( `[mergeIdentity] dropped live membership ${row.stripe_subscription_id} (${row.publication}) on merge collision; needs manual cancellation`, ); } await tx.execute(sql` update publication_memberships set identity_id = ${targetId} where identity_id = ${sourceId} `);
// stripe_wallets: one per identity; target's wallet and saved card win. await tx.execute(sql` delete from stripe_wallets where identity_id = ${sourceId} and exists (select 1 from stripe_wallets where identity_id = ${targetId}) `); await tx.execute(sql` update stripe_wallets set identity_id = ${targetId} where identity_id = ${sourceId} `);
// stripe_connected_customers: target wins on (identity, account) // collision. A membership moved from source still carries source's // customer id on its own row, so dropping the mapping doesn't orphan the // subscription. await tx.execute(sql` delete from stripe_connected_customers s where s.identity_id = ${sourceId} and exists ( select 1 from stripe_connected_customers t where t.identity_id = ${targetId} and t.stripe_account_id = s.stripe_account_id ) `); await tx.execute(sql` update stripe_connected_customers set identity_id = ${targetId} where identity_id = ${sourceId} `);
// stripe_connected_accounts: one per identity; target wins. await tx.execute(sql` delete from stripe_connected_accounts where identity_id = ${sourceId} and exists (select 1 from stripe_connected_accounts where identity_id = ${targetId}) `); await tx.execute(sql` update stripe_connected_accounts set identity_id = ${targetId} where identity_id = ${sourceId} `);
// identities.email is unique — step via NULL so we don't collide // mid-swap. custom_domains.identity is nullable and cascades on update; // we re-set it explicitly after the final value lands because NULL→value // cascades don't fire. await tx .update(identities) .set({ email: null }) .where(eq(identities.id, targetId)); await tx .update(identities) .set({ email: null }) .where(eq(identities.id, sourceId)); await tx .update(identities) .set({ email: sourceEmail }) .where(eq(identities.id, targetId)); await tx .update(custom_domains) .set({ identity: sourceEmail }) .where(eq(custom_domains.identity_id, targetId));
await tx.delete(identities).where(eq(identities.id, sourceId)); }); } catch (e) { console.error("[mergeIdentity] transaction failed:", e); return Err("database_error"); } finally { client.release(); }
await backfillAtprotoSubscriptionsForIdentity(targetId, targetAtpDid!);
return Ok(null);}