Something went wrong. Try again.
a tool for shared writing and social publishing
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162import { 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 "app/(app)/lish/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));
// 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);}