From 0bc74a2e86379958fb2c10b7d569e6b2e23532fd Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Tue, 28 Apr 2026 23:02:06 -0400 Subject: [PATCH] add merge account flow if user subs with other method --- actions/mergeIdentity.ts | 150 ++------------------ actions/publications/subscribeEmail.tsx | 70 ++++++++- app/api/oauth/[route]/route.ts | 63 +++++++-- components/Subscribe/HandleSubscribe.tsx | 51 +++++-- components/Subscribe/LinkIdentityModal.tsx | 69 +++++++++ components/Subscribe/SubscribeButton.tsx | 79 ++++++++--- src/mergeIdentity.ts | 156 +++++++++++++++++++++ 7 files changed, 460 insertions(+), 178 deletions(-) create mode 100644 components/Subscribe/LinkIdentityModal.tsx create mode 100644 src/mergeIdentity.ts diff --git a/actions/mergeIdentity.ts b/actions/mergeIdentity.ts index 2bb76aa4..d8713fa0 100644 --- a/actions/mergeIdentity.ts +++ b/actions/mergeIdentity.ts @@ -1,18 +1,6 @@ "use server"; import { cookies } from "next/headers"; -import { 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 { supabaseServerClient } from "supabase/serverClient"; import { AUTH_TOKEN_COOKIE, @@ -22,13 +10,10 @@ import { setAuthToken, } from "src/auth"; import { Err, Ok, type Result } from "src/result"; - -type MergeError = - | "merge_not_pending" - | "invalid_source" - | "invalid_target" - | "same_identity" - | "database_error"; +import { + mergeEmailIdentityIntoAtpIdentity, + type MergeError, +} from "src/mergeIdentity"; export async function confirmIdentityMerge(): Promise> { const jar = await cookies(); @@ -48,128 +33,11 @@ export async function confirmIdentityMerge(): Promise> return Err("invalid_source"); if (!target.identity.atp_did) return Err("invalid_target"); - const sourceId = source.identity.id; - const targetId = target.identity.id; - const sourceEmail = source.identity.email; - - const client = await pool.connect(); - try { - const db = drizzle(client); - await db.transaction(async (tx) => { - // Re-verify invariants under a row lock. Protects against racing - // concurrent merges or an identity being mutated between cookie - // resolution and this transaction. - 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"); - - // email_auth_tokens: caller is about to swap auth_token to the pending - // token (which already points at target). Source's tokens are now stale. - 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(); - } + const result = await mergeEmailIdentityIntoAtpIdentity({ + sourceId: source.identity.id, + targetId: target.identity.id, + }); + if (!result.ok) return result; await setAuthToken(pendingTokenId!); await removePendingMergeToken(); diff --git a/actions/publications/subscribeEmail.tsx b/actions/publications/subscribeEmail.tsx index ee091346..158abc37 100644 --- a/actions/publications/subscribeEmail.tsx +++ b/actions/publications/subscribeEmail.tsx @@ -1,6 +1,7 @@ "use server"; import { getIdentityData } from "actions/getIdentityData"; +import { mergeEmailIdentityIntoAtpIdentity } from "src/mergeIdentity"; import { supabaseServerClient } from "supabase/serverClient"; import { setAuthToken } from "src/auth"; import { PubConfirmEmail } from "emails/pubConfirmEmail"; @@ -29,7 +30,11 @@ type RequestError = | "suppression_delete_failed" | ConfirmationError; type RequestSuccess = { confirmed: boolean }; -type ConfirmError = "subscriber_not_found" | ConfirmationError; +type ConfirmError = + | "subscriber_not_found" + | "link_invalid_state" + | "email_belongs_to_other_account" + | ConfirmationError; type UnsubscribeError = "unauthorized" | "not_subscribed" | "database_error"; export async function requestPublicationEmailSubscription( @@ -174,6 +179,7 @@ export async function confirmPublicationEmailSubscription( publicationUri: string, emailRaw: string, code: string, + linkToCurrent: boolean = false, ): Promise> { const email = emailRaw.trim().toLowerCase(); @@ -192,8 +198,15 @@ export async function confirmPublicationEmailSubscription( // The confirmation code proves ownership of `email`. Issue (or look up) an // auth token for it so the subscriber is logged in and can one-click // subscribe to other publications from the same device. - const identityId = await ensureAuthTokenForEmail(email); - if (!identityId) return Err("database_error"); + let identityId: string | null; + if (linkToCurrent) { + const linkResult = await linkEmailToCurrentIdentity(email); + if (!linkResult.ok) return linkResult; + identityId = linkResult.value; + } else { + identityId = await ensureAuthTokenForEmail(email); + if (!identityId) return Err("database_error"); + } const [{ error: updateError }, { error: eventError }] = await Promise.all([ supabaseServerClient @@ -312,6 +325,57 @@ export async function unsubscribeFromPublication( return Ok(null); } +// Confirms an email subscription where the user has already chosen to link +// the email to their currently signed-in atp-only identity (via the +// LinkIdentityModal in the subscribe flow). Either attaches `email` to the +// current identity, or — if another email-only identity already owns it — +// merges that identity into the current one. +async function linkEmailToCurrentIdentity( + email: string, +): Promise> { + const [current, { data: existing }] = await Promise.all([ + getIdentityData(), + supabaseServerClient + .from("identities") + .select("id, atp_did") + .eq("email", email) + .maybeSingle(), + ]); + if (!current || !current.atp_did) return Err("link_invalid_state"); + + // Already linked — confirmation is a no-op for the identity row itself. + if (current.email && current.email.toLowerCase() === email) + return Ok(current.id); + + // Current identity already has a *different* email. Linking would clobber + // it, which the modal didn't promise — refuse. + if (current.email) return Err("link_invalid_state"); + + if (!existing || existing.id === current.id) { + const { error } = await supabaseServerClient + .from("identities") + .update({ email }) + .eq("id", current.id); + if (error) { + console.error("[subscribeEmail] attach email failed:", error); + return Err("database_error"); + } + return Ok(current.id); + } + + // Existing identity owns this email and isn't us. We can only merge if it's + // an unlinked email-only account; otherwise it has its own atp_did and + // merging would silently drop one of the two Bluesky links. + if (existing.atp_did) return Err("email_belongs_to_other_account"); + + const merged = await mergeEmailIdentityIntoAtpIdentity({ + sourceId: existing.id, + targetId: current.id, + }); + if (!merged.ok) return Err("database_error"); + return Ok(current.id); +} + async function ensureAuthTokenForEmail(email: string): Promise { const existingIdentity = await getIdentityData(); if (existingIdentity) return existingIdentity.id; diff --git a/app/api/oauth/[route]/route.ts b/app/api/oauth/[route]/route.ts index aa72fa0c..d5a8c95e 100644 --- a/app/api/oauth/[route]/route.ts +++ b/app/api/oauth/[route]/route.ts @@ -16,11 +16,16 @@ import { parseActionFromSearchParam, } from "./afterSignInActions"; import { inngest } from "app/api/inngest/client"; +import { mergeEmailIdentityIntoAtpIdentity } from "src/mergeIdentity"; type OauthRequestClientState = { redirect: string | null; action: ActionAfterSignIn | null; link?: boolean; + // Auto-confirm a cross-identity merge instead of routing to /merge-accounts. + // Set when the caller already showed an in-context "link this account?" + // confirmation (e.g. the subscribe-flow LinkIdentityModal). + autoMerge?: boolean; }; export async function GET( @@ -39,11 +44,17 @@ export async function GET( const handle = searchParams.get("handle") as string; const signup = searchParams.get("signup") === "true"; const link = searchParams.get("link") === "true"; + const autoMerge = searchParams.get("autoMerge") === "true"; // Put originating page here! let redirect = searchParams.get("redirect_url"); if (redirect) redirect = decodeURIComponent(redirect); let action = parseActionFromSearchParam(searchParams.get("action")); - let state: OauthRequestClientState = { redirect, action, link }; + let state: OauthRequestClientState = { + redirect, + action, + link, + autoMerge, + }; // Revoke any pending authentication requests if the connection is closed (optional) const ac = new AbortController(); @@ -90,7 +101,8 @@ export async function GET( // Explicit link flow from the LoginModal for an email-only user. Never // fall through to a normal DID login — we must either attach the atp_did - // to the existing email identity or route to /merge-accounts. + // to the existing email identity or route to /merge-accounts (or merge + // inline if autoMerge is set). if ( s.link && currentIdentity && @@ -105,9 +117,27 @@ export async function GET( return handleAction(s.action, redirectPath); } if (identity.id !== currentIdentity.id) { - await stagePendingMerge(identity.id, redirectPath); - // Only reached if the token insert failed. Fall through to the - // normal sign-in flow rather than blocking the user. + if (s.autoMerge) { + const merged = await mergeEmailIdentityIntoAtpIdentity({ + sourceId: currentIdentity.id, + targetId: identity.id, + }); + if (merged.ok) { + // Source identity is gone; clear it so the cross-identity + // merge block below doesn't try to merge a deleted row. + currentIdentity = null; + } else { + console.error( + "[oauth/callback] autoMerge failed:", + merged.error, + ); + await stagePendingMerge(identity.id, redirectPath); + } + } else { + await stagePendingMerge(identity.id, redirectPath); + // Only reached if the token insert failed. Fall through to the + // normal sign-in flow rather than blocking the user. + } } // Same identity already linked — fall through to refresh session. } @@ -133,10 +163,25 @@ export async function GET( currentIdentity.email ) { // DID already has an identity row. Caller is currently signed in as a - // *different* email-only identity. Stage a pending merge and let the - // user confirm on /merge-accounts before we touch either account. - await stagePendingMerge(identity.id, redirectPath); - // Only reached if the token insert failed — fall through. + // *different* email-only identity. Either merge inline (autoMerge + // means the caller already collected the user's confirmation) or + // stage a pending merge for /merge-accounts. + if (s.autoMerge) { + const merged = await mergeEmailIdentityIntoAtpIdentity({ + sourceId: currentIdentity.id, + targetId: identity.id, + }); + if (!merged.ok) { + console.error( + "[oauth/callback] autoMerge failed:", + merged.error, + ); + await stagePendingMerge(identity.id, redirectPath); + } + } else { + await stagePendingMerge(identity.id, redirectPath); + // Only reached if the token insert failed — fall through. + } } // Trigger migration if identity needs it diff --git a/components/Subscribe/HandleSubscribe.tsx b/components/Subscribe/HandleSubscribe.tsx index cb11cca7..be798e08 100644 --- a/components/Subscribe/HandleSubscribe.tsx +++ b/components/Subscribe/HandleSubscribe.tsx @@ -13,6 +13,7 @@ import { HandleInput } from "./HandleInput"; import { Avatar } from "components/Avatar"; import { useIdentityData } from "components/IdentityProvider"; import { useRecordFromDid } from "src/utils/useRecordFromDid"; +import { LinkIdentityModal } from "./LinkIdentityModal"; const apps = [ { name: "Leaflet", logo: "/logos/leaflet.svg" }, { name: "Bluesky", logo: "/logos/bluesky.svg" }, @@ -54,6 +55,26 @@ export const SubscribeWithHandle = (props: { let [loading, setLoading] = useState(false); let [subscribing, setSubscribing] = useState(false); let [oauthError, setOauthError] = useState(null); + // When an email-only user subscribes via the atproto flow, we surface a + // confirmation modal first ("link Bluesky to your account?") so they can't + // accidentally orphan their email account. + let [pendingLinkHandle, setPendingLinkHandle] = useState(null); + const viewerEmail = identity?.email; + const viewerAtpDid = identity?.atp_did; + const needsLinkConfirmation = + !!viewerEmail && !viewerAtpDid; + + const redirectToOauthForSubscribe = (handle: string, link: boolean) => { + let action = encodeActionToSearchParam({ + action: "subscribe", + publication: props.publicationUri, + }); + let url = new URL(window.location.href); + url.searchParams.set("refreshAuth", ""); + let redirectUrl = encodeURIComponent(url.toString()); + let extra = link ? "&link=true&autoMerge=true" : ""; + window.location.href = `/api/oauth/login?handle=${encodeURIComponent(handle)}&redirect_url=${redirectUrl}&action=${action}${extra}`; + }; if (props.user.loggedIn && props.user.handle) { return ( @@ -119,15 +140,12 @@ export const SubscribeWithHandle = (props: { onSubmit={(handle) => { let trimmed = handle.trim(); if (!trimmed) return; + if (needsLinkConfirmation) { + setPendingLinkHandle(trimmed); + return; + } setLoading(true); - let action = encodeActionToSearchParam({ - action: "subscribe", - publication: props.publicationUri, - }); - let url = new URL(window.location.href); - url.searchParams.set("refreshAuth", ""); - let redirectUrl = encodeURIComponent(url.toString()); - window.location.href = `/api/oauth/login?handle=${encodeURIComponent(trimmed)}&redirect_url=${redirectUrl}&action=${action}`; + redirectToOauthForSubscribe(trimmed, false); }} action=
Subscribe @@ -136,6 +154,23 @@ export const SubscribeWithHandle = (props: {
+ {needsLinkConfirmation && ( + { + if (!open) setPendingLinkHandle(null); + }} + signedInAs={viewerEmail!} + linkingIdentity={`@${pendingLinkHandle ?? ""}`} + confirmButtonLabel="Link Bluesky" + confirming={loading} + onConfirm={() => { + if (!pendingLinkHandle) return; + setLoading(true); + redirectToOauthForSubscribe(pendingLinkHandle, true); + }} + /> + )}
); }; diff --git a/components/Subscribe/LinkIdentityModal.tsx b/components/Subscribe/LinkIdentityModal.tsx new file mode 100644 index 00000000..73a5923c --- /dev/null +++ b/components/Subscribe/LinkIdentityModal.tsx @@ -0,0 +1,69 @@ +"use client"; +import { useState } from "react"; +import { useRouter } from "next/navigation"; +import { mutate } from "swr"; +import { ButtonPrimary, ButtonSecondary } from "components/Buttons"; +import { Modal } from "components/Modal"; +import { DotLoader } from "components/utils/DotLoader"; + +export const LinkIdentityModal = (props: { + open: boolean; + onOpenChange: (open: boolean) => void; + signedInAs: string; + linkingIdentity: string; + confirmButtonLabel: string; + onConfirm: () => void | Promise; + confirming?: boolean; +}) => { + let router = useRouter(); + let [loggingOut, setLoggingOut] = useState(false); + + return ( + { + if (loggingOut || props.confirming) return; + props.onOpenChange(o); + }} + className="w-[400px] max-w-full" + > +
+

+ You're signed in as{" "} + {props.signedInAs} +

+

+ Do you want to link{" "} + {props.linkingIdentity} to this + account? +

+
+ { + if (loggingOut) return; + setLoggingOut(true); + try { + await fetch("/api/auth/logout"); + } finally { + mutate("identity", null); + router.refresh(); + } + }} + > + {loggingOut ? : "Log out"} + + { + if (props.confirming) return; + void props.onConfirm(); + }} + > + {props.confirming ? : props.confirmButtonLabel} + +
+
+ + ); +}; diff --git a/components/Subscribe/SubscribeButton.tsx b/components/Subscribe/SubscribeButton.tsx index ce83f951..a4f81008 100644 --- a/components/Subscribe/SubscribeButton.tsx +++ b/components/Subscribe/SubscribeButton.tsx @@ -4,10 +4,12 @@ import { useRouter } from "next/navigation"; import { SubscribeWithHandle } from "./HandleSubscribe"; import { EmailInput, EmailConfirm } from "./EmailSubscribe"; import { EmailSubscribeSuccess } from "./EmailSubscribeSuccess"; +import { LinkIdentityModal } from "./LinkIdentityModal"; import { Modal } from "components/Modal"; import { ButtonPrimary } from "components/Buttons"; import { ManageSubscription } from "./ManageSubscribe"; import { useToaster } from "components/Toast"; +import { useIdentityData } from "components/IdentityProvider"; import { requestPublicationEmailSubscription, confirmPublicationEmailSubscription, @@ -73,6 +75,7 @@ export const SubscribeInput = (props: SubscribeProps) => { let toaster = useToaster(); let router = useRouter(); const user = useViewerSubscription(props.publicationUri); + const { identity } = useIdentityData(); let [email, setEmail] = useState(user.email ?? ""); let [confirmState, setConfirmState] = useState<"confirm" | "success">( "confirm", @@ -81,6 +84,39 @@ export const SubscribeInput = (props: SubscribeProps) => { let [requesting, setRequesting] = useState(false); let [confirming, setConfirming] = useState(false); let [locallySubscribed, setLocallySubscribed] = useState(false); + let [linkModalOpen, setLinkModalOpen] = useState(false); + // Tracks that the user passed through LinkIdentityModal — when they enter + // the confirmation code we attach the email to their current atp identity + // (or merge from any existing email-only identity) instead of creating a + // disconnected email-only account. + let [linkToCurrent, setLinkToCurrent] = useState(false); + + const viewerHandle = identity?.bsky_profiles?.handle; + const viewerAtpDid = identity?.atp_did; + const viewerEmail = identity?.email; + // The atp-only-but-subscribing-via-email case: signed in as a Bluesky + // account with no email yet. The modal asks them to link the typed email + // (or log out) before we send a confirmation code. + const needsLinkConfirmation = !!viewerAtpDid && !viewerEmail && !!email; + + const sendRequest = async (link: boolean) => { + setRequesting(true); + setLinkToCurrent(link); + let res = await requestPublicationEmailSubscription( + props.publicationUri, + email, + ); + setRequesting(false); + if (!res.ok) { + toaster({ type: "error", content: ERROR_MESSAGES[res.error] }); + return; + } + if (res.value.confirmed) { + setConfirmState("success"); + router.refresh(); + } + setConfirmOpen(true); + }; const isSubscribed = user.subscribed || locallySubscribed; return ( @@ -106,24 +142,11 @@ export const SubscribeInput = (props: SubscribeProps) => { disabled={requesting || !email} onClick={async () => { if (requesting) return; - setRequesting(true); - let res = await requestPublicationEmailSubscription( - props.publicationUri, - email, - ); - setRequesting(false); - if (!res.ok) { - toaster({ - type: "error", - content: ERROR_MESSAGES[res.error], - }); + if (needsLinkConfirmation) { + setLinkModalOpen(true); return; } - if (res.value.confirmed) { - setConfirmState("success"); - router.refresh(); - } - setConfirmOpen(true); + await sendRequest(false); }} > Subscribe @@ -138,6 +161,20 @@ export const SubscribeInput = (props: SubscribeProps) => { onSubscribed={() => setLocallySubscribed(true)} /> )} + {props.newsletterMode && needsLinkConfirmation && ( + { + setLinkModalOpen(false); + await sendRequest(true); + }} + /> + )} {props.newsletterMode && ( { if (!open) { if (confirmState === "success") setLocallySubscribed(true); setConfirmState("confirm"); + setLinkToCurrent(false); } }} > @@ -164,6 +202,7 @@ export const SubscribeInput = (props: SubscribeProps) => { props.publicationUri, email, code, + linkToCurrent, ); setConfirming(false); if (!res.ok) { @@ -192,7 +231,9 @@ type SubscribeError = | "invalid_code" | "database_error" | "suppressed_spam_complaint" - | "suppression_delete_failed"; + | "suppression_delete_failed" + | "link_invalid_state" + | "email_belongs_to_other_account"; const ERROR_MESSAGES: Record = { invalid_email: "Please enter a valid email address.", @@ -205,4 +246,8 @@ const ERROR_MESSAGES: Record = { "This address was previously marked as spam and can't be resubscribed. Contact the publication to resolve.", suppression_delete_failed: "We couldn't clear a prior delivery issue on this address. Try again later.", + link_invalid_state: + "Couldn't link this email to your account. Try logging out and subscribing again.", + email_belongs_to_other_account: + "This email is already linked to a different Bluesky account. Log out to use that account instead.", }; diff --git a/src/mergeIdentity.ts b/src/mergeIdentity.ts new file mode 100644 index 00000000..e941a3a9 --- /dev/null +++ b/src/mergeIdentity.ts @@ -0,0 +1,156 @@ +import { 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"; + +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> { + const { sourceId, targetId } = args; + if (sourceId === targetId) return Err("same_identity"); + + const client = await pool.connect(); + 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"); + + 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(); + } + + return Ok(null); +} -- 2.51.2