diff --git a/actions/emailAuth.tsx b/actions/emailAuth.tsx
index 50e27163..bbe8e636 100644
--- a/actions/emailAuth.tsx
+++ b/actions/emailAuth.tsx
@@ -7,7 +7,7 @@ import { email_auth_tokens, identities } from "drizzle/schema";
import { and, desc, eq, gt } from "drizzle-orm";
import { setAuthToken } from "src/auth";
import { postAuthRedirect } from "src/postAuthRedirect";
-import { applyAfterSignInAction } from "src/emailSubscription";
+import { applyAfterSignInAction } from "src/subscriptions/email";
import { pool } from "supabase/pool";
import { supabaseServerClient } from "supabase/serverClient";
import { LeafletConfirmEmail } from "emails/leafletConfirmEmail";
diff --git a/actions/publications/joinMembership.ts b/actions/publications/joinMembership.ts
index 7f8b08f9..ace1ca89 100644
--- a/actions/publications/joinMembership.ts
+++ b/actions/publications/joinMembership.ts
@@ -13,11 +13,7 @@ import {
} from "stripe/wallet";
import { isActiveMembership, filterJoinableTiers } from "src/membership";
import { getReaderMembership, notifyNewMember } from "src/membership.server";
-import { publishAtprotoSubscriptionForDid } from "actions/publications/subscribeToPublication";
-import {
- checkEmailSubscriptionAllowed,
- recordEmailSubscription,
-} from "src/emailSubscription";
+import { ensureSubscriberRecordsForMembership } from "src/subscriptions/membership";
import {
sanitizeSubscriptionSource,
type SubscriptionSource,
@@ -346,44 +342,13 @@ export async function subscribeToTier(args: {
} catch (e) {
console.error("[joinMembership] new member notification failed:", e);
}
- // A member is also a subscriber: record an email subscription when the
- // publication sends newsletters — recordEmailSubscription also mirrors
- // the join onto the reader's PDS. Skipped when a confirmed subscription
- // already exists (the sign-in-during-join paths subscribe before
- // payment), so the subscriber events and analytics aren't doubled. When
- // the email path doesn't run — already subscribed, newsletter disabled,
- // suppressed address, or a failure — mirror the PDS record directly so
- // atproto readers still see the subscription (it dedupes internally).
// Awaited (not after()) so the client's identity refetch on success
- // sees the new rows. Best-effort — the join already billed, so a
- // subscriber-mirroring failure shouldn't fail it.
- try {
- const [{ data: existingEmailSub }, allowed] = await Promise.all([
- supabaseServerClient
- .from("publication_email_subscribers")
- .select("state")
- .eq("publication", args.publicationUri)
- .eq("email", identity.email)
- .maybeSingle(),
- checkEmailSubscriptionAllowed(args.publicationUri, identity.email),
- ]);
- const recorded =
- allowed.ok && existingEmailSub?.state !== "confirmed"
- ? await recordEmailSubscription(
- args.publicationUri,
- identity.email,
- identity.id,
- sanitizeSubscriptionSource(args.source),
- )
- : null;
- if (!recorded?.ok && identity.atp_did)
- await publishAtprotoSubscriptionForDid(
- identity.atp_did,
- args.publicationUri,
- );
- } catch (e) {
- console.error("[joinMembership] subscriber mirroring failed:", e);
- }
+ // sees the new rows.
+ await ensureSubscriberRecordsForMembership(
+ args.publicationUri,
+ { id: identity.id, email: identity.email, atp_did: identity.atp_did },
+ sanitizeSubscriptionSource(args.source),
+ );
}
// Concurrent joins (different tier/cadence, so different idempotency keys)
diff --git a/actions/publications/subscribeEmail.tsx b/actions/publications/subscribeEmail.tsx
index 382a0fee..c531e460 100644
--- a/actions/publications/subscribeEmail.tsx
+++ b/actions/publications/subscribeEmail.tsx
@@ -6,7 +6,10 @@ import { supabaseServerClient } from "supabase/serverClient";
import {
recordEmailSubscription,
checkEmailSubscriptionAllowed,
-} from "src/emailSubscription";
+ disableEmailSubscription,
+ upsertSubscriber,
+ onEmailSubscriptionConfirmed,
+} from "src/subscriptions/email";
import { PubConfirmEmail } from "emails/pubConfirmEmail";
import { Ok, Err, type Result } from "src/result";
import {
@@ -16,18 +19,15 @@ import {
sendConfirmationEmail,
type ConfirmationError,
} from "src/utils/confirmationEmail";
+import { backfillAtprotoSubscriptionsForIdentity } from "src/subscriptions/atproto";
import {
- backfillAtprotoSubscriptionsForIdentity,
- publishAtprotoSubscriptionForDid,
- unsubscribeToPublication,
-} from "actions/publications/subscribeToPublication";
-import type { OAuthSessionError } from "src/atproto-oauth";
+ fullUnsubscribe,
+ type FullUnsubscribeError,
+} from "src/subscriptions/unsubscribe";
import { normalizePublicationRecord } from "src/utils/normalizeRecords";
import { linkOrphanedEmailSubscribers } from "src/utils/linkOrphanedEmailSubscribers";
import { blobRefToSrc, EMAIL_ICON_TRANSFORM } from "src/utils/blobRefToSrc";
import { AtUri } from "@atproto/api";
-import { after } from "next/server";
-import { trackSubscriptionEvent } from "src/subscriptionAnalytics";
import {
sanitizeSubscriptionSource,
type SubscriptionSource,
@@ -45,11 +45,7 @@ type ConfirmError =
| "link_invalid_state"
| "email_belongs_to_other_account"
| ConfirmationError;
-type UnsubscribeError =
- | "unauthorized"
- | "not_subscribed"
- | "database_error"
- | OAuthSessionError;
+type UnsubscribeError = "unauthorized" | FullUnsubscribeError;
export async function requestPublicationEmailSubscription(
publicationUri: string,
@@ -95,54 +91,16 @@ export async function requestPublicationEmailSubscription(
return Ok({ confirmed: true });
}
- // A previous `unsubscribed` row makes the first event `resubscribed`.
const pendingCode = generateConfirmationCode();
- const { data: existing } = await supabaseServerClient
- .from("publication_email_subscribers")
- .select("state")
- .eq("publication", publicationUri)
- .eq("email", email)
- .maybeSingle();
- const wasUnsubscribed = existing?.state === "unsubscribed";
-
- const { data: subscriber, error } = await supabaseServerClient
- .from("publication_email_subscribers")
- .upsert(
- {
- publication: publicationUri,
- email,
- identity_id: identity?.id ?? null,
- state: "pending",
- confirmation_code: pendingCode,
- unsubscribed_at: null,
- },
- { onConflict: "publication,email" },
- )
- .select("id")
- .single();
- if (error || !subscriber) {
- console.error("[subscribeEmail] upsert subscriber failed:", error);
- return Err("database_error");
- }
-
- const { error: eventsError } = await supabaseServerClient
- .from("publication_email_subscriber_events")
- .insert([
- {
- subscriber: subscriber.id,
- publication: publicationUri,
- event_type: wasUnsubscribed ? "resubscribed" : "subscribe_requested",
- },
- {
- subscriber: subscriber.id,
- publication: publicationUri,
- event_type: "confirmation_sent",
- },
- ]);
- if (eventsError) {
- console.error("[subscribeEmail] insert events failed:", eventsError);
- return Err("database_error");
- }
+ const upserted = await upsertSubscriber({
+ publicationUri,
+ email,
+ identityId: identity?.id ?? null,
+ state: "pending",
+ confirmationCode: pendingCode,
+ event: "confirmation_sent",
+ });
+ if (!upserted.ok) return Err(upserted.error);
const sent = await sendConfirmationEmail({
to: email,
@@ -221,124 +179,63 @@ export async function confirmPublicationEmailSubscription(
return Err("database_error");
}
- const { data: confirmedIdentity } = await supabaseServerClient
- .from("identities")
- .select("atp_did")
- .eq("id", identityId)
- .maybeSingle();
-
- // The atproto mirror below is the same subscription, not a second one — only
- // the email event is tracked.
- after(() =>
- trackSubscriptionEvent({
- event: "subscribe",
- method: "email",
- origin: "app",
- publicationUri,
- subscriberDid: confirmedIdentity?.atp_did,
- subscriberEmail: email,
- source: sanitizeSubscriptionSource(source),
- }),
+ await onEmailSubscriptionConfirmed(
+ publicationUri,
+ email,
+ identityId,
+ sanitizeSubscriptionSource(source),
);
-
- if (confirmedIdentity?.atp_did) {
- await publishAtprotoSubscriptionForDid(
- confirmedIdentity.atp_did,
- publicationUri,
- );
- }
-
return Ok(null);
}
+// A viewer is "subscribed" if they have an email subscription, an atproto
+// subscription, or a membership (see useViewerSubscription). Unsubscribe
+// clears all of them — except an active paid membership, which the reader must
+// cancel first (or, when it's already set to cancel at period end, confirm
+// forfeiting the remaining time via `cancelPaidImmediately`).
export async function unsubscribeFromPublication(
publicationUri: string,
+ opts?: { cancelPaidImmediately?: boolean },
): Promise> {
const identity = await getAuthIdentity();
if (!identity?.id) return Err("unauthorized");
- // A viewer is "subscribed" if they have EITHER an email subscription OR an
- // atproto subscription (see useViewerSubscription). Unsubscribe must clear
- // both — otherwise a user with only an atproto sub hits "not_subscribed"
- // and a user with both ends up half-unsubscribed.
- const [{ data: subscribers }, { data: atprotoSub }] = await Promise.all([
- supabaseServerClient
- .from("publication_email_subscribers")
- .select("id, state")
- .eq("publication", publicationUri)
- .eq("identity_id", identity.id),
- identity.atp_did
- ? supabaseServerClient
- .from("publication_subscriptions")
- .select("uri")
- .eq("identity", identity.atp_did)
- .eq("publication", publicationUri)
- .maybeSingle()
- : Promise.resolve({ data: null }),
- ]);
- // Identity may have multiple matching subscriber rows historically
- // (different emails tied to the same identity, e.g. a linked address that
- // predates an email change). Unsubscribe all of them at once — a single
- // "Unsubscribe" click from this user should stop all future newsletter
- // mail from this publication.
- const active = (subscribers ?? []).filter((s) => s.state !== "unsubscribed");
-
- if (active.length === 0 && !atprotoSub) {
- if (subscribers && subscribers.length > 0) return Ok(null); // already unsubscribed; idempotent
- return Err("not_subscribed");
- }
-
- if (active.length > 0) {
- const ids = active.map((s) => s.id);
- const nowIso = new Date().toISOString();
- const [{ error: updateError }, { error: eventError }] = await Promise.all([
- supabaseServerClient
- .from("publication_email_subscribers")
- .update({
- state: "unsubscribed",
- unsubscribed_at: nowIso,
- confirmation_code: null,
- })
- .in("id", ids),
- supabaseServerClient.from("publication_email_subscriber_events").insert(
- active.map((s) => ({
- subscriber: s.id,
- publication: publicationUri,
- event_type: "unsubscribe_requested",
- })),
- ),
- ]);
- if (updateError || eventError) {
- console.error(
- "[subscribeEmail] unsubscribe update/event failed:",
- updateError ?? eventError,
- );
- return Err("database_error");
- }
- // The atproto half (if any) is tracked inside unsubscribeToPublication.
- after(() =>
- trackSubscriptionEvent({
- event: "unsubscribe",
- method: "email",
- origin: "app",
- publicationUri,
- subscriberDid: identity.atp_did,
- subscriberEmail: identity.email,
- }),
- );
- }
+ return fullUnsubscribe({
+ publication: publicationUri,
+ identity,
+ cancelPaidImmediately: opts?.cancelPaidImmediately,
+ allowStripeCancel: true,
+ });
+}
- if (atprotoSub) {
- const atprotoResult = await unsubscribeToPublication(publicationUri);
- if (!atprotoResult.success) return Err(atprotoResult.error);
- }
+// The email-notifications toggle in the manage-subscription panel. Turning
+// email off only mutes delivery (the atproto record and any membership keep
+// the subscription alive); turning it on runs the resubscribe path with the
+// identity's own email.
+export async function setEmailNotifications(
+ publicationUri: string,
+ enabled: boolean,
+): Promise<
+ Result<
+ null,
+ | "unauthorized"
+ | "no_email"
+ | "database_error"
+ | "newsletter_disabled"
+ | "suppressed_spam_complaint"
+ | "suppression_delete_failed"
+ >
+> {
+ const identity = await getAuthIdentity();
+ if (!identity?.id) return Err("unauthorized");
- // NOTE: Postmark Suppressions API is deliberately NOT called here. Per spec,
- // in-app unsubscribes only flip local state — this prevents a Pub A
- // unsubscribe from silently breaking Pub B deliveries on the shared broadcast
- // stream. Phase 7 handles webhook-driven suppression reconciliation.
+ if (!enabled) return disableEmailSubscription(publicationUri, identity);
- return Ok(null);
+ const email = identity.email?.toLowerCase();
+ if (!email) return Err("no_email");
+ const allowed = await checkEmailSubscriptionAllowed(publicationUri, email);
+ if (!allowed.ok) return Err(allowed.error);
+ return recordEmailSubscription(publicationUri, email, identity.id);
}
// Confirms an email subscription where the user has already chosen to link
diff --git a/actions/publications/subscribeToPublication.ts b/actions/publications/subscribeToPublication.ts
index 83725071..d2987dcb 100644
--- a/actions/publications/subscribeToPublication.ts
+++ b/actions/publications/subscribeToPublication.ts
@@ -1,11 +1,7 @@
"use server";
-import { AtpBaseClient } from "lexicons/api";
import { getAuthIdentity } from "src/auth";
-import { restoreOAuthSession, OAuthSessionError } from "src/atproto-oauth";
-import { TID } from "@atproto/common";
-import { supabaseServerClient } from "supabase/serverClient";
-import { AtUri } from "@atproto/syntax";
+import { OAuthSessionError } from "src/atproto-oauth";
import { redirect } from "next/navigation";
import { headers } from "next/headers";
import { after } from "next/server";
@@ -16,11 +12,7 @@ import {
} from "src/subscriptionSource";
import { buildOauthLoginUrl } from "src/utils/customDomain";
import { encodeActionToSearchParam } from "app/api/oauth/[route]/afterSignInActions";
-import {
- Notification,
- pingIdentityToUpdateNotification,
-} from "src/notifications";
-import { v7 } from "uuid";
+import { createAtprotoSubscription } from "src/subscriptions/atproto";
type SubscribeResult =
| { success: true }
@@ -54,214 +46,27 @@ export async function subscribeToPublication(
);
}
- const sessionResult = await restoreOAuthSession(identity.atp_did);
- if (!sessionResult.ok) {
- return { success: false, error: sessionResult.error };
- }
- let credentialSession = sessionResult.value;
- let agent = new AtpBaseClient(
- credentialSession.fetchHandler.bind(credentialSession),
- );
-
- let { data: existingSubscription } = await supabaseServerClient
- .from("publication_subscriptions")
- .select("uri")
- .eq("identity", credentialSession.did!)
- .eq("publication", publication)
- .maybeSingle();
- if (existingSubscription) {
- return { success: true };
- }
-
- let record = await agent.site.standard.graph.subscription.create(
- { repo: credentialSession.did!, rkey: TID.nextStr() },
- {
- publication,
- },
- );
- let { error } = await supabaseServerClient
- .from("publication_subscriptions")
- .insert({
- uri: record.uri,
- record,
- publication,
- identity: credentialSession.did!,
- });
-
- after(() =>
- trackSubscriptionEvent({
- event: "subscribe",
- method: "atproto",
- origin: "app",
- publicationUri: publication,
- subscriberDid: credentialSession.did,
- recordUri: record.uri,
- source: subscribeSource,
- }),
+ const created = await createAtprotoSubscription(
+ identity.atp_did,
+ publication,
);
-
- // Create notification for the publication owner
- let publicationOwner = new AtUri(publication).host;
- if (publicationOwner !== credentialSession.did) {
- let notification: Notification = {
- id: v7(),
- recipient: publicationOwner,
- data: {
- type: "subscribe",
- subscription_uri: record.uri,
- },
- };
- await supabaseServerClient.from("notifications").insert(notification);
- await pingIdentityToUpdateNotification(publicationOwner);
- }
-
- return {
- success: true,
- };
-}
-
-// Best-effort publish of an AT Protocol subscription record for a known DID.
-// Used by the email-subscribe flow when the subscribing email is linked to an
-// atp_did: we want the user's PDS record to match their email subscription.
-// Swallows all failures — the email subscription is the source of truth and
-// must succeed regardless of whether the atproto write goes through.
-export async function publishAtprotoSubscriptionForDid(
- atp_did: string,
- publication: string,
-): Promise {
- try {
- let { data: existingSubscription } = await supabaseServerClient
- .from("publication_subscriptions")
- .select("uri")
- .eq("identity", atp_did)
- .eq("publication", publication)
- .maybeSingle();
- if (existingSubscription) return;
-
- const sessionResult = await restoreOAuthSession(atp_did);
- if (!sessionResult.ok) return;
- let credentialSession = sessionResult.value;
- let agent = new AtpBaseClient(
- credentialSession.fetchHandler.bind(credentialSession),
- );
-
- let record = await agent.site.standard.graph.subscription.create(
- { repo: atp_did, rkey: TID.nextStr() },
- { publication },
- );
- await supabaseServerClient.from("publication_subscriptions").insert({
- uri: record.uri,
- record,
- publication,
- identity: atp_did,
- });
-
- let publicationOwner = new AtUri(publication).host;
- if (publicationOwner !== atp_did) {
- let notification: Notification = {
- id: v7(),
- recipient: publicationOwner,
- data: {
- type: "subscribe",
- subscription_uri: record.uri,
- },
- };
- await supabaseServerClient.from("notifications").insert(notification);
- await pingIdentityToUpdateNotification(publicationOwner);
- }
- } catch (e) {
- console.error(
- "[publishAtprotoSubscriptionForDid] failed:",
- atp_did,
- publication,
- e,
+ if (!created.ok) return { success: false, error: created.error };
+ // Null when a subscription already existed, which isn't a new subscribe.
+ if (created.value) {
+ let recordUri = created.value.uri;
+ let subscriberDid = identity.atp_did;
+ after(() =>
+ trackSubscriptionEvent({
+ event: "subscribe",
+ method: "atproto",
+ origin: "app",
+ publicationUri: publication,
+ subscriberDid,
+ recordUri,
+ source: subscribeSource,
+ }),
);
}
-}
-
-// Publishes atproto subscription records for every confirmed email
-// subscription tied to `identityId`. Call after an identity gains an atp_did
-// (account-linking, merge) so existing email-only subscriptions also live as
-// records on the user's PDS — otherwise they'd be invisible to atproto-side
-// consumers and would silently drop if a publication ever leaves email-only
-// mode. Best-effort: errors are swallowed by publishAtprotoSubscriptionForDid.
-export async function backfillAtprotoSubscriptionsForIdentity(
- identityId: string,
- atp_did: string,
-): Promise {
- const { data: subs } = await supabaseServerClient
- .from("publication_email_subscribers")
- .select("publication")
- .eq("identity_id", identityId)
- .eq("state", "confirmed");
- if (!subs || subs.length === 0) return;
-
- await Promise.all(
- subs.map((s) => publishAtprotoSubscriptionForDid(atp_did, s.publication)),
- );
-}
-type UnsubscribeResult =
- | { success: true }
- | { success: false; error: OAuthSessionError };
-
-export async function unsubscribeToPublication(
- publication: string,
-): Promise {
- let identity = await getAuthIdentity();
- if (!identity || !identity.atp_did) {
- return {
- success: false,
- error: {
- type: "oauth_session_expired",
- message: "Not authenticated",
- did: "",
- },
- };
- }
-
- const sessionResult = await restoreOAuthSession(identity.atp_did);
- if (!sessionResult.ok) {
- return { success: false, error: sessionResult.error };
- }
- let credentialSession = sessionResult.value;
- let agent = new AtpBaseClient(
- credentialSession.fetchHandler.bind(credentialSession),
- );
- let { data: existingSubscription } = await supabaseServerClient
- .from("publication_subscriptions")
- .select("*")
- .eq("identity", identity.atp_did)
- .eq("publication", publication)
- .single();
- if (!existingSubscription) return { success: true };
-
- // Delete from both collections (old and new schema) - one or both may exist
- let rkey = new AtUri(existingSubscription.uri).rkey;
- await Promise.all([
- agent.pub.leaflet.graph.subscription
- .delete({ repo: credentialSession.did!, rkey })
- .catch(() => {}),
- agent.site.standard.graph.subscription
- .delete({ repo: credentialSession.did!, rkey })
- .catch(() => {}),
- ]);
-
- await supabaseServerClient
- .from("publication_subscriptions")
- .delete()
- .eq("identity", identity.atp_did)
- .eq("publication", publication);
-
- after(() =>
- trackSubscriptionEvent({
- event: "unsubscribe",
- method: "atproto",
- origin: "app",
- publicationUri: publication,
- subscriberDid: identity.atp_did,
- recordUri: existingSubscription.uri,
- }),
- );
return { success: true };
}
diff --git a/actions/reader/getSubscriptions.ts b/actions/reader/getSubscriptions.ts
index f24f68fb..54cf35e5 100644
--- a/actions/reader/getSubscriptions.ts
+++ b/actions/reader/getSubscriptions.ts
@@ -68,6 +68,41 @@ export async function getSubscriptions(
};
}
+// Subscriptions that exist only as confirmed email-subscriber rows — no
+// atproto record to show up in getSubscriptions. Always scoped to the signed-in
+// viewer (never keyed by a caller-supplied did): email subscriptions are
+// private, so they must never surface on public profile subscription lists.
+export async function getEmailOnlySubscriptions(): Promise<
+ PublicationSubscription[]
+> {
+ const identity = await getAuthIdentity();
+ if (!identity) return [];
+
+ const [{ data: emailRows }, { data: atprotoRows }] = await Promise.all([
+ supabaseServerClient
+ .from("publication_email_subscribers")
+ .select("publication")
+ .eq("identity_id", identity.id)
+ .eq("state", "confirmed"),
+ identity.atp_did
+ ? supabaseServerClient
+ .from("publication_subscriptions")
+ .select("publication")
+ .eq("identity", identity.atp_did)
+ : { data: [] },
+ ]);
+
+ // Anything the viewer also holds an atproto subscription for already renders
+ // in the main list; the same pub can also appear on multiple subscriber rows
+ // (merged identities with several historical emails), which the Set folds.
+ const atprotoPubs = new Set((atprotoRows ?? []).map((s) => s.publication));
+ return getPublicationsByUris(
+ [...new Set((emailRows ?? []).map((r) => r.publication))].filter(
+ (uri) => !atprotoPubs.has(uri),
+ ),
+ );
+}
+
// Full listing data for specific publications, in the same shape as
// getSubscriptions — used to pin paid memberships regardless of where they
// fall in the paginated subscription list.
diff --git a/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/SubscriptionsPageContent.tsx b/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/SubscriptionsPageContent.tsx
index 72389507..327923c8 100644
--- a/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/SubscriptionsPageContent.tsx
+++ b/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/SubscriptionsPageContent.tsx
@@ -16,6 +16,7 @@ export function SubscriptionsPageContent(props: {
memberships: MyMembership[];
paidPubs: PublicationSubscription[];
subscriptions: PublicationSubscription[];
+ emailOnlySubscriptions: PublicationSubscription[];
nextCursor: Cursor | null;
}) {
let pubsByUri = new Map(props.paidPubs.map((pub) => [pub.uri, pub]));
@@ -67,6 +68,7 @@ export function SubscriptionsPageContent(props: {
did={props.did}
subscriptions={props.subscriptions}
nextCursor={props.nextCursor}
+ prependSubscriptions={props.emailOnlySubscriptions}
excludeUris={paidUris}
/>
diff --git a/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/page.tsx b/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/page.tsx
index 40be4bf0..0bb2eab9 100644
--- a/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/page.tsx
+++ b/app/(app)/(identity)/(home-pages)/(writer)/subscriptions/page.tsx
@@ -4,6 +4,7 @@ import { getMyMemberships } from "actions/memberships";
import {
getSubscriptions,
getPublicationsByUris,
+ getEmailOnlySubscriptions,
} from "actions/reader/getSubscriptions";
import { SubscriptionsPageContent } from "./SubscriptionsPageContent";
@@ -11,8 +12,9 @@ export default async function SubscriptionsPage() {
const identity = await getIdentityData();
if (!identity) redirect("/home");
- const [subs, membershipsData] = await Promise.all([
+ const [subs, emailOnlySubscriptions, membershipsData] = await Promise.all([
getSubscriptions(identity.atp_did),
+ getEmailOnlySubscriptions(),
getMyMemberships(),
]);
const memberships = membershipsData?.memberships ?? [];
@@ -26,6 +28,7 @@ export default async function SubscriptionsPage() {
memberships={memberships}
paidPubs={paidPubs}
subscriptions={subs.subscriptions}
+ emailOnlySubscriptions={emailOnlySubscriptions}
nextCursor={subs.nextCursor}
/>
);
diff --git a/app/(app)/(identity)/(home-pages)/p/[didOrHandle]/subscriptions/SubscriptionsContent.tsx b/app/(app)/(identity)/(home-pages)/p/[didOrHandle]/subscriptions/SubscriptionsContent.tsx
index 63a25fca..03d8c924 100644
--- a/app/(app)/(identity)/(home-pages)/p/[didOrHandle]/subscriptions/SubscriptionsContent.tsx
+++ b/app/(app)/(identity)/(home-pages)/p/[didOrHandle]/subscriptions/SubscriptionsContent.tsx
@@ -14,7 +14,10 @@ export const ProfileSubscriptionsContent = (props: {
did: string;
subscriptions: PublicationSubscription[];
nextCursor: Cursor | null;
-
+ // Extra subscriptions merged ahead of the paginated atproto list — the
+ // viewer's own email-only subscriptions on the "Your Subscriptions" page.
+ // The public profile page never passes these.
+ prependSubscriptions?: PublicationSubscription[];
excludeUris?: string[];
}) => {
const getKey = (
@@ -73,9 +76,16 @@ export const ProfileSubscriptionsContent = (props: {
return () => observer.disconnect();
}, [data, size, setSize, isValidating]);
- const subscriptions = (
- data ? data.flatMap((page) => page.subscriptions) : []
- ).filter((sub) => !props.excludeUris?.includes(sub.uri));
+ const seenUris = new Set();
+ const subscriptions = [
+ ...(props.prependSubscriptions ?? []),
+ ...(data ? data.flatMap((page) => page.subscriptions) : []),
+ ].filter((sub) => {
+ if (props.excludeUris?.includes(sub.uri) || seenUris.has(sub.uri))
+ return false;
+ seenUris.add(sub.uri);
+ return true;
+ });
if (subscriptions.length === 0 && !isValidating) {
return ;
diff --git a/app/api/auth/email-login/route.ts b/app/api/auth/email-login/route.ts
index 576756c4..72abcda7 100644
--- a/app/api/auth/email-login/route.ts
+++ b/app/api/auth/email-login/route.ts
@@ -4,7 +4,7 @@ import { NextRequest, NextResponse } from "next/server";
import { AUTH_TOKEN_COOKIE, resolveAuthToken } from "src/auth";
import { postAuthRedirect } from "src/postAuthRedirect";
import { publicationUriForHost } from "src/utils/publicationForHost";
-import { applyAfterSignInAction } from "src/emailSubscription";
+import { applyAfterSignInAction } from "src/subscriptions/email";
import { parseActionFromSearchParam } from "app/api/oauth/[route]/afterSignInActions";
import { supabaseServerClient } from "supabase/serverClient";
import { normalizePublicationRecord } from "src/utils/normalizeRecords";
diff --git a/app/api/inngest/functions/send_post_broadcast.ts b/app/api/inngest/functions/send_post_broadcast.ts
index a9fb336d..640191c4 100644
--- a/app/api/inngest/functions/send_post_broadcast.ts
+++ b/app/api/inngest/functions/send_post_broadcast.ts
@@ -17,6 +17,7 @@ import {
import { PubLeafletPagesLinearDocument } from "lexicons/api";
import type { AppBskyFeedDefs } from "@atproto/api";
import { hydrateBskyPostBlocks } from "src/utils/fetchBskyPosts";
+import { manageSubscriptionUrl } from "src/subscriptions/manageUrl";
import { fetchStandardSiteBlockData } from "src/utils/fetchStandardSiteBlockData";
import { getProfiles } from "src/identity";
import {
@@ -41,6 +42,8 @@ const BATCH_SIZE = 500;
// so we only pay the React Email render cost once per batch.
const UNSUB_PLACEHOLDER =
"https://placeholder.leaflet.pub/unsubscribe-token-replace-me";
+const MANAGE_PLACEHOLDER =
+ "https://placeholder.leaflet.pub/manage-subscription-replace-me";
export const send_post_broadcast = inngest.createFunction(
{
@@ -346,6 +349,7 @@ export const send_post_broadcast = inngest.createFunction(
did,
assetsBaseUrl: `${assetsBaseUrl}/`,
unsubscribeUrl: UNSUB_PLACEHOLDER,
+ manageUrl: MANAGE_PLACEHOLDER,
membersUpsell: group.upsell ? membersUpsell : undefined,
}),
);
@@ -373,9 +377,16 @@ export const send_post_broadcast = inngest.createFunction(
const unsubscribeUrl = `${assetsBaseUrl}/emails/unsubscribe?unsubscribe_token=${encodeURIComponent(
sub.unsubscribe_token,
)}`;
+ const manageUrl = manageSubscriptionUrl({
+ baseUrl: assetsBaseUrl,
+ email: sub.email,
+ publicationUrl: pubProps.publicationUrl,
+ });
const htmlBody = htmlTemplate
.split(UNSUB_PLACEHOLDER)
- .join(unsubscribeUrl);
+ .join(unsubscribeUrl)
+ .split(MANAGE_PLACEHOLDER)
+ .join(manageUrl.replace(/&/g, "&"));
return {
MessageStream: "broadcast",
From: fromHeader,
diff --git a/app/api/oauth/[route]/route.ts b/app/api/oauth/[route]/route.ts
index e88fbb92..784bbaac 100644
--- a/app/api/oauth/[route]/route.ts
+++ b/app/api/oauth/[route]/route.ts
@@ -1,7 +1,5 @@
-import {
- backfillAtprotoSubscriptionsForIdentity,
- subscribeToPublication,
-} from "actions/publications/subscribeToPublication";
+import { subscribeToPublication } from "actions/publications/subscribeToPublication";
+import { backfillAtprotoSubscriptionsForIdentity } from "src/subscriptions/atproto";
import { recommendAction } from "actions/recommendAction";
import { cookies } from "next/headers";
import { redirect } from "next/navigation";
diff --git a/app/api/webhooks/stripe/connect-events/handlers.ts b/app/api/webhooks/stripe/connect-events/handlers.ts
index 36fac9fb..b30be81a 100644
--- a/app/api/webhooks/stripe/connect-events/handlers.ts
+++ b/app/api/webhooks/stripe/connect-events/handlers.ts
@@ -3,6 +3,7 @@ import { render } from "@react-email/render";
import { getStripe } from "stripe/client";
import { supabaseServerClient } from "supabase/serverClient";
import { notifyNewMember } from "src/membership.server";
+import { ensureSubscriberRecordsForMembership } from "src/subscriptions/membership";
import MembershipPaymentFailed from "emails/membershipPaymentFailed";
const APP_URL = process.env.NEXT_PUBLIC_APP_URL || "https://leaflet.pub";
@@ -15,6 +16,23 @@ function isActiveStatus(status: string | null | undefined): boolean {
return status === "active" || status === "trialing";
}
+// A membership activated by webhook needs the same subscriber mirroring the
+// inline join flow does.
+async function mirrorSubscriberRecords(
+ identityId: string,
+ publication: string,
+) {
+ const { data: identity } = await supabaseServerClient
+ .from("identities")
+ .select("id, email, atp_did")
+ .eq("id", identityId)
+ .maybeSingle();
+ if (!identity) return;
+ await ensureSubscriberRecordsForMembership(publication, identity, null, {
+ respectUnsubscribed: true,
+ });
+}
+
// Reconcile a membership row from a connected-account subscription event. Keyed
// on the subscription id; events for subscriptions with no matching row fall
// through to reconcileUntrackedSubscription, which can rebuild the row from the
@@ -66,6 +84,11 @@ export async function handleMembershipSubscriptionEvent(
if (isActiveStatus(sub.status) && !wasActive) {
await notifyNewMember(sub.metadata.publication, existing.id);
+ if (sub.metadata.publication && sub.metadata.identity_id)
+ await mirrorSubscriberRecords(
+ sub.metadata.identity_id,
+ sub.metadata.publication,
+ );
}
}
@@ -142,6 +165,7 @@ async function reconcileUntrackedSubscription(
.eq("id", tracked.id);
if (error) throw error;
await notifyNewMember(publication, tracked.id);
+ await mirrorSubscriberRecords(identity_id, publication);
return;
}
@@ -200,6 +224,7 @@ async function reconcileUntrackedSubscription(
if (isActiveStatus(sub.status)) {
await notifyNewMember(publication, inserted.id);
+ await mirrorSubscriberRecords(identity_id, publication);
}
}
diff --git a/app/emails/unsubscribe/route.ts b/app/emails/unsubscribe/route.ts
index e3cc6f1c..f682e03b 100644
--- a/app/emails/unsubscribe/route.ts
+++ b/app/emails/unsubscribe/route.ts
@@ -1,72 +1,79 @@
import { NextRequest, NextResponse } from "next/server";
-import { after } from "next/server";
import { supabaseServerClient } from "supabase/serverClient";
-import { trackSubscriptionEvent } from "src/subscriptionAnalytics";
+import { fullUnsubscribe } from "src/subscriptions/unsubscribe";
+import { disableEmailSubscription } from "src/subscriptions/email";
+import { manageSubscriptionUrl } from "src/subscriptions/manageUrl";
+import { normalizePublicationRecord } from "src/utils/normalizeRecords";
+import { MAIN_SITE_URL } from "src/utils/customDomain";
-// Looks up a subscriber by unsubscribe_token and flips their state to
-// `unsubscribed`. Idempotent — already-unsubscribed rows return ok without
-// appending a duplicate event.
+// Looks up a subscriber by unsubscribe_token and fully unsubscribes them:
+// email rows flip to `unsubscribed` and, for a linked atproto account, the
+// subscription record is removed too. The one exception is an active paid
+// membership — a token link must never touch Stripe, so those readers get
+// email delivery turned off and a pointer to the signed-in manage flow, with
+// the membership (and its atproto record) intact. Idempotent —
+// already-unsubscribed rows return ok without appending a duplicate event.
//
// NOTE: Postmark Suppressions API is intentionally NOT called here. This
// endpoint is hit from in-email footer links AND from Postmark's one-click
// List-Unsubscribe-Post — but for BOTH cases, Phase 7 handles webhook-driven
// suppression reconciliation. The immediate effect of this route is purely
// local state + event log.
-async function unsubscribeByToken(
- token: string,
-): Promise<{ ok: boolean; email?: string; publicationName?: string }> {
+
+type UnsubscribeOutcome =
+ | { result: "invalid" }
+ | {
+ result: "unsubscribed" | "membership_active" | "error";
+ email: string;
+ publicationName?: string;
+ manageUrl: string;
+ };
+
+async function unsubscribeByToken(token: string): Promise {
const { data: sub } = await supabaseServerClient
.from("publication_email_subscribers")
- .select("id, email, state, publication, publications(record)")
+ .select("id, email, state, publication, identity_id, publications(record)")
.eq("unsubscribe_token", token)
.maybeSingle();
- if (!sub) return { ok: false };
-
- const publicationName = (() => {
- const rec = sub.publications?.record as
- | { name?: string }
- | null
- | undefined;
- return rec?.name;
- })();
-
- if (sub.state === "unsubscribed") {
- return { ok: true, email: sub.email, publicationName };
- }
+ if (!sub) return { result: "invalid" };
- const nowIso = new Date().toISOString();
- const [{ error: updateError }, { error: eventError }] = await Promise.all([
- supabaseServerClient
- .from("publication_email_subscribers")
- .update({
- state: "unsubscribed",
- unsubscribed_at: nowIso,
- confirmation_code: null,
- })
- .eq("id", sub.id),
- supabaseServerClient.from("publication_email_subscriber_events").insert({
- subscriber: sub.id,
- publication: sub.publication,
- event_type: "unsubscribe_requested",
- }),
- ]);
- if (updateError || eventError) {
- console.error(
- "[unsubscribe] update/event failed:",
- updateError ?? eventError,
- );
- return { ok: false };
+ const normalized = normalizePublicationRecord(sub.publications?.record);
+ const publicationName = normalized?.name;
+ const manageUrl = manageSubscriptionUrl({
+ baseUrl: MAIN_SITE_URL,
+ email: sub.email,
+ publicationUrl: normalized?.url,
+ });
+ const base = { email: sub.email, publicationName, manageUrl };
+
+ const { data: identity } = sub.identity_id
+ ? await supabaseServerClient
+ .from("identities")
+ .select("id, atp_did, email")
+ .eq("id", sub.identity_id)
+ .maybeSingle()
+ : { data: null };
+
+ const result = await fullUnsubscribe({
+ publication: sub.publication,
+ identity,
+ subscriberRows: [{ id: sub.id, state: sub.state }],
+ });
+ if (result.ok) return { result: "unsubscribed", ...base };
+
+ // A membership error implies fullUnsubscribe resolved one, which it only
+ // does for a non-null identity.
+ if (
+ identity &&
+ (result.error === "membership_active" ||
+ result.error === "membership_pending_cancellation")
+ ) {
+ const disabled = await disableEmailSubscription(sub.publication, identity);
+ return disabled.ok
+ ? { result: "membership_active", ...base }
+ : { result: "error", ...base };
}
- after(() =>
- trackSubscriptionEvent({
- event: "unsubscribe",
- method: "email",
- origin: "app",
- publicationUri: sub.publication,
- subscriberEmail: sub.email,
- }),
- );
- return { ok: true, email: sub.email, publicationName };
+ return { result: "error", ...base };
}
// Escape HTML-unsafe chars for interpolation into a static page body.
@@ -110,6 +117,7 @@ function htmlPage(title: string, bodyHtml: string): string {
h1 { font-size: 20px; margin: 0 0 12px; }
p { margin: 0 0 8px; line-height: 1.5; color: #555; }
.email { font-weight: 600; color: #272727; }
+ .manage { display: inline-block; margin-top: 12px; color: #272727; }
@@ -121,6 +129,17 @@ function htmlPage(title: string, bodyHtml: string): string {
`;
}
+function page(status: number, title: string, bodyHtml: string): NextResponse {
+ return new NextResponse(htmlPage(title, bodyHtml), {
+ status,
+ headers: { "Content-Type": "text/html; charset=utf-8" },
+ });
+}
+
+function manageLink(manageUrl: string): string {
+ return `Manage subscription `;
+}
+
export async function POST(request: NextRequest) {
const token = request.nextUrl.searchParams.get("unsubscribe_token");
// Postmark fires List-Unsubscribe-Post with no body. Always return 200 —
@@ -132,37 +151,45 @@ export async function POST(request: NextRequest) {
export async function GET(request: NextRequest) {
const token = request.nextUrl.searchParams.get("unsubscribe_token");
- if (!token) {
- return new NextResponse(
- htmlPage(
- "Unsubscribe link invalid",
- `This unsubscribe link is missing a token. Try opening it directly from the email footer.
`,
- ),
- { status: 400, headers: { "Content-Type": "text/html; charset=utf-8" } },
+ if (!token)
+ return page(
+ 400,
+ "Unsubscribe link invalid",
+ `This unsubscribe link is missing a token. Try opening it directly from the email footer.
`,
);
- }
- const result = await unsubscribeByToken(token);
- if (!result.ok) {
- return new NextResponse(
- htmlPage(
- "Unsubscribe link invalid",
- `We couldn't find this subscription. It may have already been removed.
`,
- ),
- {
- status: 404,
- headers: { "Content-Type": "text/html; charset=utf-8" },
- },
+
+ const outcome = await unsubscribeByToken(token);
+ if (outcome.result === "invalid")
+ return page(
+ 404,
+ "Unsubscribe link invalid",
+ `We couldn't find this subscription. It may have already been removed.
`,
);
- }
- const emailLine = result.email
- ? `Future emails to ${esc(result.email)} from ${
- result.publicationName
- ? esc(result.publicationName)
- : "this publication"
- } have been stopped.
`
- : `Your subscription has been removed.
`;
- return new NextResponse(
- htmlPage("Unsubscribed", emailLine),
- { status: 200, headers: { "Content-Type": "text/html; charset=utf-8" } },
+
+ const pubName = outcome.publicationName
+ ? esc(outcome.publicationName)
+ : "this publication";
+ const stopped = `Future emails to ${esc(outcome.email)} from ${pubName} have been stopped`;
+
+ if (outcome.result === "error")
+ return page(
+ 500,
+ "Something went wrong",
+ `
We couldn't complete the unsubscribe. Please try again, or manage your subscription directly.
+ ${manageLink(outcome.manageUrl)}`,
+ );
+ if (outcome.result === "membership_active")
+ return page(
+ 200,
+ "Emails stopped",
+ `${stopped}.
+ Your paid membership is still active — you can change or cancel it from your subscription settings.
+ ${manageLink(outcome.manageUrl)}`,
+ );
+ return page(
+ 200,
+ "Unsubscribed",
+ `${stopped}, and your subscription has been removed.
+ ${manageLink(outcome.manageUrl)}`,
);
}
diff --git a/app/test/subscribe/page.tsx b/app/test/subscribe/page.tsx
index eace2547..c8435c9b 100644
--- a/app/test/subscribe/page.tsx
+++ b/app/test/subscribe/page.tsx
@@ -38,9 +38,12 @@ function makeIdentity(opts: {
atp_did: opts.hasHandle ? profile.did : null,
email: opts.hasEmail ? "reader@example.com" : null,
bsky_profiles: opts.hasHandle ? { ...profile, record: profile } : null,
- publication_subscriptions: opts.subscribed
- ? [{ publication: PUBLICATION_URI }]
- : [],
+ // An atproto subscription implies an atproto account — gate on the handle
+ // so the unified `subscribed` derivation sees a coherent identity.
+ publication_subscriptions:
+ opts.subscribed && opts.hasHandle
+ ? [{ publication: PUBLICATION_URI }]
+ : [],
publication_email_subscribers:
opts.subscribed && opts.hasEmail
? [{ publication: PUBLICATION_URI, state: "confirmed" }]
diff --git a/components/Memberships/JoinMembershipFlow.tsx b/components/Memberships/JoinMembershipFlow.tsx
index 7176e47c..4ab44bc2 100644
--- a/components/Memberships/JoinMembershipFlow.tsx
+++ b/components/Memberships/JoinMembershipFlow.tsx
@@ -139,9 +139,7 @@ export function JoinMembershipFlow(props: {
const effectiveCadence = (tier: Tier): Cadence =>
tier.annual_price_cents != null ? cadence : "month";
- const isSubscribed = props.newsletterMode
- ? viewerSub.emailSubscribed
- : viewerSub.atprotoSubscribed;
+ const isSubscribed = viewerSub.subscribed;
const hasNeededIdentity = props.newsletterMode
? !!identity?.email
: !!identity?.bsky_profiles?.handle;
diff --git a/components/Subscribe/ManageSubscribe.tsx b/components/Subscribe/ManageSubscribe.tsx
index d47c9c92..efa3ffc8 100644
--- a/components/Subscribe/ManageSubscribe.tsx
+++ b/components/Subscribe/ManageSubscribe.tsx
@@ -1,14 +1,14 @@
"use client";
import { refreshIdentityData } from "components/IdentityProvider";
-import { useState } from "react";
+import { useEffect, useState } from "react";
import { useRouter } from "next/navigation";
import { ButtonPrimary, ButtonSecondary } from "components/Buttons";
import { CheckTiny } from "components/Icons/CheckTiny";
import { GoToArrow } from "components/Icons/GoToArrow";
import { Modal } from "components/Modal";
+import { Toggle } from "components/Toggle";
import { useToaster } from "components/Toast";
import { DotLoader } from "components/utils/DotLoader";
-import { isOAuthSessionError, OAuthErrorMessage } from "components/OAuthError";
import {
SwitchPlanForm,
CancelMembershipForm,
@@ -29,6 +29,7 @@ import {
requestPublicationEmailSubscription,
confirmPublicationEmailSubscription,
unsubscribeFromPublication,
+ setEmailNotifications,
} from "actions/publications/subscribeEmail";
const BLUESKY_FEED_URL =
@@ -37,17 +38,40 @@ const BLUESKY_FEED_URL =
const prefClassName =
"flex gap-2 justify-between font-bold text-secondary items-center";
+// The manage link in newsletter emails lands back on the publication with
+// ?manage_subscription=1 (after a pass through email-login) so the panel opens
+// without hunting for the button. Several surfaces on one page can mount this
+// component; stripping the param claims it, so only the first one opens.
+function readManageSubscriptionReturn(): boolean {
+ if (typeof window === "undefined") return false;
+ const params = new URLSearchParams(window.location.search);
+ if (!params.get("manage_subscription")) return false;
+ params.delete("manage_subscription");
+ const qs = params.toString();
+ window.history.replaceState(
+ null,
+ "",
+ window.location.pathname + (qs ? `?${qs}` : ""),
+ );
+ return true;
+}
+
export const ManageSubscription = (props: {
publicationUri: string;
publicationUrl?: string;
newsletterMode: boolean;
user: ViewerUser;
}) => {
+ let [open, setOpen] = useState(false);
let [membershipView, setMembershipView] = useState<{
type: "switch" | "cancel";
membership: MyMembership;
} | null>(null);
+ useEffect(() => {
+ if (readManageSubscriptionReturn()) setOpen(true);
+ }, []);
+
const closeMembershipView = () => setMembershipView(null);
const onMembershipChanged = () => {
setMembershipView(null);
@@ -57,6 +81,7 @@ export const ManageSubscription = (props: {
return (
{membershipView?.type === "switch"
@@ -67,8 +92,9 @@ export const ManageSubscription = (props: {
}
className="w-md max-w-full"
- onOpenChange={(open) => {
- if (!open) setMembershipView(null);
+ onOpenChange={(o) => {
+ setOpen(o);
+ if (!o) setMembershipView(null);
}}
// A real button rather than a div Radix wraps in one: the wrapper made the
// row 1.5px taller than the subscribe button it replaces, which is a
@@ -142,9 +168,43 @@ const ManageSubscriptionContent = (props: {
let [linkRequesting, setLinkRequesting] = useState(false);
let [linkConfirming, setLinkConfirming] = useState(false);
let [unsubscribing, setUnsubscribing] = useState(false);
+ let [confirmForfeit, setConfirmForfeit] = useState(false);
+ // Optimistic value, held from the click until the identity refetch lands (or
+ // the action fails); non-null also means a toggle is already in flight.
+ let [emailOverride, setEmailOverride] = useState(null);
let toaster = useToaster();
let router = useRouter();
+ const notify = (type: "success" | "error", message: string) =>
+ toaster({ type, content: {message}
});
+ const notifyError = (error: string, fallback: string) =>
+ notify("error", ERROR_MESSAGES[error] ?? fallback);
+ const refreshSubscription = () => {
+ refreshIdentityData();
+ router.refresh();
+ };
+
+ const emailEnabled = emailOverride ?? user.emailEnabled;
+ useEffect(() => {
+ setEmailOverride(null);
+ }, [user.emailEnabled]);
+
+ const onToggleEmail = async () => {
+ if (emailOverride !== null) return;
+ const next = !emailEnabled;
+ setEmailOverride(next);
+ const res = await setEmailNotifications(props.publicationUri, next);
+ if (!res.ok) {
+ setEmailOverride(null);
+ notifyError(
+ res.error,
+ "We couldn't update your email notifications. Please try again!",
+ );
+ return;
+ }
+ refreshSubscription();
+ };
+
const onRequestLink = async () => {
if (linkRequesting || !email) return;
setLinkRequesting(true);
@@ -154,25 +214,13 @@ const ManageSubscriptionContent = (props: {
);
setLinkRequesting(false);
if (!res.ok) {
- toaster({
- type: "error",
- content: (
-
- {LINK_ERROR_MESSAGES[res.error] ??
- "We couldn't link that email. Please try again!"}
-
- ),
- });
+ notifyError(res.error, "We couldn't link that email. Please try again!");
return;
}
if (res.value.confirmed) {
- toaster({
- content: Email Linked!
,
- type: "success",
- });
+ notify("success", "Email Linked!");
setLinkEmailOpen(false);
- refreshIdentityData();
- router.refresh();
+ refreshSubscription();
return;
}
setLinkEmailOpen(true);
@@ -188,51 +236,26 @@ const ManageSubscriptionContent = (props: {
);
setLinkConfirming(false);
if (!res.ok) {
- toaster({
- type: "error",
- content: (
-
- {LINK_ERROR_MESSAGES[res.error] ??
- "We couldn't confirm the code. Please try again!"}
-
- ),
- });
+ notifyError(res.error, "We couldn't confirm the code. Please try again!");
return;
}
- toaster({
- content: Email Linked!
,
- type: "success",
- });
+ notify("success", "Email Linked!");
setLinkEmailOpen(false);
- refreshIdentityData();
- router.refresh();
+ refreshSubscription();
};
- const onUnsubscribe = async () => {
+ const onUnsubscribe = async (opts?: { cancelPaidImmediately?: boolean }) => {
if (unsubscribing) return;
setUnsubscribing(true);
- const res = await unsubscribeFromPublication(props.publicationUri);
+ const res = await unsubscribeFromPublication(props.publicationUri, opts);
setUnsubscribing(false);
if (!res.ok) {
- toaster({
- type: "error",
- content: isOAuthSessionError(res.error) ? (
-
- ) : (
-
- {UNSUBSCRIBE_ERROR_MESSAGES[res.error] ??
- "We couldn't unsubscribe you. Please try again!"}
-
- ),
- });
+ notifyError(res.error, "We couldn't unsubscribe you. Please try again!");
return;
}
- toaster({
- content: Unsubscribed!
,
- type: "success",
- });
- refreshIdentityData();
- router.refresh();
+ notify("success", "Unsubscribed!");
+ mutateMyMembership(props.publicationUri);
+ refreshSubscription();
};
if (isLoading)
@@ -242,12 +265,22 @@ const ManageSubscriptionContent = (props: {
);
- const cancellableMembership =
- membership &&
- isMembershipActive(membership.status) &&
- !membership.cancelAtPeriodEnd
- ? membership
- : null;
+ const activeMembership =
+ membership && isMembershipActive(membership.status) ? membership : null;
+ // A membership already winding down: fully unsubscribing now cancels the
+ // Stripe subscription immediately, forfeiting the remaining paid time — so it
+ // goes through an explicit confirm. One that isn't must be cancelled first.
+ const pendingCancellation = activeMembership?.cancelAtPeriodEnd
+ ? activeMembership
+ : null;
+ // Turning email off is only a mute when something else keeps the
+ // subscription alive; for an email-only reader it *is* the unsubscribe.
+ // deriveSubscriptionState would drop `subscribed`, unmounting this panel
+ // mid-interaction with no way back in — so they get the address as a plain
+ // row (like Linked Handle below) and the deliberate Unsubscribe button.
+ // Turning email back on is always safe.
+ const canMuteEmail =
+ !emailEnabled || user.atprotoSubscribed || !!activeMembership;
return (
@@ -257,9 +290,18 @@ const ManageSubscriptionContent = (props: {
{props.newsletterMode && user.email ? (
-
Linked Email
+
Email Notifications
{user.email}
+ {canMuteEmail ? (
+
+
+ {emailEnabled
+ ? "Turn off email notifications"
+ : "Turn on email notifications"}
+
+
+ ) : null}
) : null}
{user.handle && (
@@ -338,18 +380,32 @@ const ManageSubscriptionContent = (props: {
) : null}
- {cancellableMembership ? (
-
props.onCancel(cancellableMembership)}
- >
- Cancel Membership
-
+ {activeMembership && !pendingCancellation ? (
+ <>
+
props.onCancel(activeMembership)}
+ >
+ Cancel Membership
+
+
+ Cancel your membership first to fully unsubscribe.
+
+ >
+ ) : pendingCancellation && confirmForfeit ? (
+
setConfirmForfeit(false)}
+ onConfirm={() => onUnsubscribe({ cancelPaidImmediately: true })}
+ />
) : (
+ pendingCancellation ? setConfirmForfeit(true) : onUnsubscribe()
+ }
>
{unsubscribing ? "Unsubscribing…" : "Unsubscribe"}
@@ -358,6 +414,47 @@ const ManageSubscriptionContent = (props: {
);
};
+const ForfeitConfirm = (props: {
+ membership: MyMembership;
+ unsubscribing: boolean;
+ onBack: () => void;
+ onConfirm: () => void;
+}) => {
+ const endDate = useLocalizedDate(props.membership.currentPeriodEnd ?? "", {
+ year: "numeric",
+ month: "short",
+ day: "numeric",
+ });
+ return (
+
+
+ Your membership{" "}
+ {props.membership.currentPeriodEnd ? (
+ <>
+ lasts until {endDate}
+ >
+ ) : (
+ "is still active"
+ )}
+ . Unsubscribing now cancels it immediately and forfeits the remaining
+ time.
+
+
+
+ Back
+
+
+ {props.unsubscribing ? : "Unsubscribe now"}
+
+
+
+ );
+};
+
const MembershipSection = (props: {
membership: MyMembership;
onSwitch: (membership: MyMembership) => void;
@@ -394,23 +491,25 @@ const MembershipSection = (props: {
);
};
-const LINK_ERROR_MESSAGES: Record = {
+// Every error code the panel's actions can return; the per-action fallback
+// covers anything not listed.
+const ERROR_MESSAGES: Record = {
+ unauthorized: "Sign in to manage your subscription.",
+ database_error: "Something went wrong. Please try again.",
invalid_email: "Please enter a valid email address.",
+ no_email: "Link an email to your account first.",
newsletter_disabled: "This publication isn't accepting email subscriptions.",
email_send_failed:
"We couldn't send the confirmation email. Please try again.",
subscriber_not_found: "No pending subscription. Please try again.",
invalid_code: "That code didn't match. Please try again.",
- database_error: "Something went wrong. Please try again.",
suppressed_spam_complaint:
- "This address was previously marked as spam and can't be linked. Contact the publication to resolve.",
+ "This address was previously marked as spam. Contact the publication to resolve.",
suppression_delete_failed:
"We couldn't clear a prior delivery issue on this address. Try again later.",
-};
-
-const UNSUBSCRIBE_ERROR_MESSAGES: Record = {
- unauthorized: "Sign in to manage your subscription.",
- not_subscribed:
- "We couldn't find an email subscription for this publication.",
- database_error: "Something went wrong. Try again.",
+ membership_active: "Cancel your membership first, then unsubscribe.",
+ membership_pending_cancellation:
+ "Your membership is still active — confirm cancelling it to unsubscribe.",
+ stripe_error:
+ "We couldn't cancel your membership's billing. Please try again.",
};
diff --git a/components/Subscribe/PaidSubscribeButton.tsx b/components/Subscribe/PaidSubscribeButton.tsx
index dd03fd38..1e9824c9 100644
--- a/components/Subscribe/PaidSubscribeButton.tsx
+++ b/components/Subscribe/PaidSubscribeButton.tsx
@@ -45,21 +45,27 @@ export const PaidSubscribeButton = (
setOpen(true);
}, []);
- const showManage = props.newsletterMode
- ? user.emailSubscribed
- : user.atprotoSubscribed;
-
return (
<>
-
-
+ {user.subscribed ? (
+
+ ) : (
+ setOpen(true)}
+ >
+ Subscribe
+
+ )}
{/* Stays mounted while the flow subscribes the reader mid-join, so the
- payment step isn't lost when showManage flips. */}
+ payment step isn't lost when the trigger above flips to Manage. */}
{
setConfirmOpen(true);
};
- const showManage = props.newsletterMode
- ? user.emailSubscribed
- : user.atprotoSubscribed;
- const isSubscribed = showManage || locallySubscribed;
+ const isSubscribed = user.subscribed || locallySubscribed;
const modeMenu = (
);
@@ -204,7 +201,7 @@ export const SubscribeInput = (props: SubscribeProps) => {
{props.newsletterMode &&
user.atprotoSubscribed &&
- !user.emailSubscribed ? (
+ !user.emailEnabled ? (
{
if (joinable.hasPaidTiers && joinable.tiers)
return
;
- const showManage = props.newsletterMode
- ? user.emailSubscribed
- : user.atprotoSubscribed;
-
const subscribeTrigger = (
Subscribe
@@ -389,7 +382,7 @@ export const SubscribeButton = (props: SubscribeProps) => {
return (
<>
- {showManage || locallySubscribed ? (
+ {user.subscribed || locallySubscribed ? (
{
- const atprotoSubscribed =
- !!identity &&
- (identity.publication_subscriptions ?? []).some(
- (s) => s.publication === publicationUri,
- );
- const emailSubscribed =
- !!identity &&
- (identity.publication_email_subscribers ?? []).some(
- (s) => s.publication === publicationUri && s.state === "confirmed",
- );
+ // No identity means no rows, which derives to "not subscribed".
+ const state = deriveSubscriptionState(publicationUri, {
+ subscriptions: identity?.publication_subscriptions,
+ emailSubscribers: identity?.publication_email_subscribers,
+ memberships: identity?.publication_memberships,
+ });
return {
loggedIn: !!identity,
email: identity?.email ?? undefined,
handle: identity?.bsky_profiles?.handle ?? undefined,
- atprotoSubscribed,
- emailSubscribed,
+ ...state,
};
}, [identity, publicationUri]);
}
diff --git a/emails/post.tsx b/emails/post.tsx
index a7170ea6..679b0f63 100644
--- a/emails/post.tsx
+++ b/emails/post.tsx
@@ -14,7 +14,7 @@ import {
} from "@react-email/components";
import type { PrismLanguage } from "@react-email/code-block";
import { UnicodeString, type AppBskyFeedDefs } from "@atproto/api";
-import React, { type CSSProperties } from "react";
+import React, { Fragment, type CSSProperties } from "react";
import {
PubLeafletBlocksBlockquote,
PubLeafletBlocksBskyPost,
@@ -81,6 +81,7 @@ type PostEmailProps = {
* real unsubscribe link — used by the Phase 4 preview-send path.
*/
unsubscribeUrl?: string;
+ manageUrl?: string;
/**
* Absolute URL used to resolve bundled /email-assets/* images. Postmark
* fetches ` ` verbatim so these must be absolute in outgoing mail.
@@ -935,17 +936,33 @@ export const PostEmail = (props: Partial = {}) => {
}}
>
{p.unsubscribeUrl ? (
-
- Unsubscribe
-
+ [
+ p.manageUrl && {
+ href: p.manageUrl,
+ label: "Manage subscription",
+ },
+ {
+ href: p.unsubscribeUrl,
+ label: "Unsubscribe",
+ },
+ ]
+ .filter((l) => !!l)
+ .map((l, i) => (
+
+ {i > 0 ? " · " : null}
+
+ {l.label}
+
+
+ ))
) : (
(preview — not sent to subscribers)
diff --git a/src/emailSubscription.ts b/src/emailSubscription.ts
deleted file mode 100644
index 3077c214..00000000
--- a/src/emailSubscription.ts
+++ /dev/null
@@ -1,177 +0,0 @@
-import { supabaseServerClient } from "supabase/serverClient";
-import { publishAtprotoSubscriptionForDid } from "actions/publications/subscribeToPublication";
-import { parseActionFromSearchParam } from "app/api/oauth/[route]/afterSignInActions";
-import {
- getSuppression,
- deleteSuppression,
-} from "src/utils/postmarkSuppressions";
-import { Ok, Err, type Result } from "src/result";
-import { after } from "next/server";
-import { trackSubscriptionEvent } from "src/subscriptionAnalytics";
-import {
- sanitizeSubscriptionSource,
- type SubscriptionSource,
-} from "src/subscriptionSource";
-
-export async function recordEmailSubscription(
- publicationUri: string,
- email: string,
- identityId: string,
- source?: SubscriptionSource | null,
-): Promise> {
- const { data: existing } = await supabaseServerClient
- .from("publication_email_subscribers")
- .select("state")
- .eq("publication", publicationUri)
- .eq("email", email)
- .maybeSingle();
- const wasUnsubscribed = existing?.state === "unsubscribed";
-
- const { data: subscriber, error } = await supabaseServerClient
- .from("publication_email_subscribers")
- .upsert(
- {
- publication: publicationUri,
- email,
- identity_id: identityId,
- state: "confirmed",
- confirmation_code: null,
- confirmed_at: new Date().toISOString(),
- unsubscribed_at: null,
- },
- { onConflict: "publication,email" },
- )
- .select("id")
- .single();
- if (error || !subscriber) {
- console.error("[recordEmailSubscription] upsert failed:", error);
- return Err("database_error");
- }
-
- const { error: eventsError } = await supabaseServerClient
- .from("publication_email_subscriber_events")
- .insert([
- {
- subscriber: subscriber.id,
- publication: publicationUri,
- event_type: wasUnsubscribed ? "resubscribed" : "subscribe_requested",
- },
- {
- subscriber: subscriber.id,
- publication: publicationUri,
- event_type: "confirmed",
- },
- ]);
- if (eventsError) {
- console.error("[recordEmailSubscription] events failed:", eventsError);
- return Err("database_error");
- }
-
- const { data: confirmedIdentity } = await supabaseServerClient
- .from("identities")
- .select("atp_did")
- .eq("id", identityId)
- .maybeSingle();
- // The atproto mirror below is the same subscription, not a second one — only
- // the email event is tracked.
- after(() =>
- trackSubscriptionEvent({
- event: "subscribe",
- method: "email",
- origin: "app",
- publicationUri,
- subscriberDid: confirmedIdentity?.atp_did,
- subscriberEmail: email,
- source,
- }),
- );
-
- if (confirmedIdentity?.atp_did) {
- await publishAtprotoSubscriptionForDid(
- confirmedIdentity.atp_did,
- publicationUri,
- );
- }
-
- return Ok(null);
-}
-
-export async function checkEmailSubscriptionAllowed(
- publicationUri: string,
- email: string,
-): Promise<
- Result<
- null,
- | "newsletter_disabled"
- | "suppressed_spam_complaint"
- | "suppression_delete_failed"
- >
-> {
- const { data: settings } = await supabaseServerClient
- .from("publication_newsletter_settings")
- .select("enabled")
- .eq("publication", publicationUri)
- .maybeSingle();
- if (!settings?.enabled) return Err("newsletter_disabled");
-
- const suppression = await getSuppression(email);
- if (suppression?.reason === "SpamComplaint") {
- return Err("suppressed_spam_complaint");
- }
- if (
- suppression?.reason === "HardBounce" ||
- suppression?.reason === "ManualSuppression"
- ) {
- const deleted = await deleteSuppression(email);
- if (!deleted) return Err("suppression_delete_failed");
- }
- return Ok(null);
-}
-
-// Runs the after-sign-in action (parsed from the `action` search param) for a
-// just-authenticated email identity and returns where to send the browser. When
-// there is no action — or no authenticated identity — the redirect is returned
-// unchanged; only a recognized action does extra work. `subscribe` is currently
-// the only handler. Called from the email-login flow's two mutually-exclusive
-// entry points: the reused-session fast path in app/api/auth/email-login/route.ts
-// and the fresh-confirm path in confirmEmailLogin.
-export async function applyAfterSignInAction(
- action: string | null,
- redirect: string,
- email: string | null,
- identityId: string | null,
-): Promise {
- const parsed = parseActionFromSearchParam(action);
- if (!parsed || !email || !identityId) return redirect;
-
- switch (parsed.action) {
- case "subscribe": {
- const target = new URL(redirect);
- const allowed = await checkEmailSubscriptionAllowed(
- parsed.publication,
- email,
- );
- if (!allowed.ok) {
- target.searchParams.set("subscribe_email_error", allowed.error);
- return target.toString();
- }
- const recorded = await recordEmailSubscription(
- parsed.publication,
- email,
- identityId,
- sanitizeSubscriptionSource(parsed.source),
- );
- if (!recorded.ok) {
- target.searchParams.set("subscribe_email_error", recorded.error);
- return target.toString();
- }
- // Paid-membership joins carry ?join_tier in the redirect itself, so the
- // reader resumes payment where they left off rather than detouring here.
- target.searchParams.set("subscribe_email", email);
- target.searchParams.set("subscribed_pub", parsed.publication);
- return target.toString();
- }
- default:
- return redirect;
- }
-}
diff --git a/src/membership.server.ts b/src/membership.server.ts
index 2dd903c8..6597d622 100644
--- a/src/membership.server.ts
+++ b/src/membership.server.ts
@@ -136,7 +136,7 @@ export async function getReaderMembership(
const { data } = await supabaseServerClient
.from("publication_memberships")
.select(
- "id, tier, status, current_period_end, stripe_customer_id, stripe_subscription_id, stripe_account_id, cadence",
+ "id, tier, status, current_period_end, cancel_at_period_end, stripe_customer_id, stripe_subscription_id, stripe_account_id, cadence",
)
.eq("publication", publicationUri)
.eq("identity_id", identityId)
diff --git a/src/mergeIdentity.ts b/src/mergeIdentity.ts
index 09444705..55c4733d 100644
--- a/src/mergeIdentity.ts
+++ b/src/mergeIdentity.ts
@@ -11,7 +11,7 @@ import {
} from "drizzle/schema";
import { pool } from "supabase/pool";
import { Err, Ok, type Result } from "src/result";
-import { backfillAtprotoSubscriptionsForIdentity } from "actions/publications/subscribeToPublication";
+import { backfillAtprotoSubscriptionsForIdentity } from "src/subscriptions/atproto";
export type MergeError =
| "merge_not_pending"
diff --git a/src/subscriptions/atproto.ts b/src/subscriptions/atproto.ts
new file mode 100644
index 00000000..29eb5299
--- /dev/null
+++ b/src/subscriptions/atproto.ts
@@ -0,0 +1,170 @@
+import { AtpBaseClient } from "lexicons/api";
+import { restoreOAuthSession, type OAuthSessionError } from "src/atproto-oauth";
+import { TID } from "@atproto/common";
+import { supabaseServerClient } from "supabase/serverClient";
+import { AtUri } from "@atproto/syntax";
+import { Ok, Err, type Result } from "src/result";
+import { trackSubscriptionEvent } from "src/subscriptionAnalytics";
+import {
+ Notification,
+ pingIdentityToUpdateNotification,
+} from "src/notifications";
+import { v7 } from "uuid";
+
+// The PDS-record half of a subscription. Plain lib — these take arbitrary DIDs
+// and perform PDS writes via the server-held OAuth sessions, so they must
+// never be exported from a "use server" file (that would make them
+// wire-callable with any DID).
+
+// Writes the subscription record to the DID's PDS, mirrors it into
+// publication_subscriptions, and notifies the publication owner. Idempotent:
+// an existing row short-circuits to Ok(null), so no caller has to check first.
+// Throws on a failed PDS write — subscribeToPublication surfaces that to the
+// reader; publishAtprotoSubscriptionForDid is the best-effort wrapper.
+export async function createAtprotoSubscription(
+ atp_did: string,
+ publication: string,
+): Promise> {
+ let { data: existingSubscription } = await supabaseServerClient
+ .from("publication_subscriptions")
+ .select("uri")
+ .eq("identity", atp_did)
+ .eq("publication", publication)
+ .maybeSingle();
+ if (existingSubscription) return Ok(null);
+
+ const sessionResult = await restoreOAuthSession(atp_did);
+ if (!sessionResult.ok) return Err(sessionResult.error);
+ let credentialSession = sessionResult.value;
+ let agent = new AtpBaseClient(
+ credentialSession.fetchHandler.bind(credentialSession),
+ );
+
+ let record = await agent.site.standard.graph.subscription.create(
+ { repo: atp_did, rkey: TID.nextStr() },
+ { publication },
+ );
+ await supabaseServerClient.from("publication_subscriptions").insert({
+ uri: record.uri,
+ record,
+ publication,
+ identity: atp_did,
+ });
+
+ let publicationOwner = new AtUri(publication).host;
+ if (publicationOwner !== atp_did) {
+ let notification: Notification = {
+ id: v7(),
+ recipient: publicationOwner,
+ data: {
+ type: "subscribe",
+ subscription_uri: record.uri,
+ },
+ };
+ await supabaseServerClient.from("notifications").insert(notification);
+ await pingIdentityToUpdateNotification(publicationOwner);
+ }
+ return Ok({ uri: record.uri });
+}
+
+// Best-effort mirror for flows where the subscription already exists elsewhere
+// (an email subscriber row, a paid membership) and the PDS record is a copy:
+// those flows are the source of truth and must succeed whether or not the
+// atproto write goes through.
+export async function publishAtprotoSubscriptionForDid(
+ atp_did: string,
+ publication: string,
+): Promise {
+ try {
+ await createAtprotoSubscription(atp_did, publication);
+ } catch (e) {
+ console.error(
+ "[publishAtprotoSubscriptionForDid] failed:",
+ atp_did,
+ publication,
+ e,
+ );
+ }
+}
+
+// Publishes atproto subscription records for every confirmed email
+// subscription tied to `identityId`. Call after an identity gains an atp_did
+// (account-linking, merge) so existing email-only subscriptions also live as
+// records on the user's PDS — otherwise they'd be invisible to atproto-side
+// consumers and would silently drop if a publication ever leaves email-only
+// mode. Best-effort: errors are swallowed by publishAtprotoSubscriptionForDid.
+export async function backfillAtprotoSubscriptionsForIdentity(
+ identityId: string,
+ atp_did: string,
+): Promise {
+ const { data: subs } = await supabaseServerClient
+ .from("publication_email_subscribers")
+ .select("publication")
+ .eq("identity_id", identityId)
+ .eq("state", "confirmed");
+ if (!subs || subs.length === 0) return;
+
+ await Promise.all(
+ subs.map((s) => publishAtprotoSubscriptionForDid(atp_did, s.publication)),
+ );
+}
+
+// Best-effort removal of a DID's subscription record + row. The DB row delete
+// is the effective part (it drives rosters, counts, and the viewer's
+// subscribed state); a PDS delete blocked by a dead OAuth session just leaves
+// a stale record behind, which we tolerate the same way we tolerate records
+// deleted out-of-band from another client.
+export async function deleteAtprotoSubscriptionForDid(
+ atp_did: string,
+ publication: string,
+): Promise {
+ try {
+ let { data: existingSubscription } = await supabaseServerClient
+ .from("publication_subscriptions")
+ .select("uri")
+ .eq("identity", atp_did)
+ .eq("publication", publication)
+ .maybeSingle();
+ if (!existingSubscription) return;
+
+ const sessionResult = await restoreOAuthSession(atp_did);
+ if (sessionResult.ok) {
+ let credentialSession = sessionResult.value;
+ let agent = new AtpBaseClient(
+ credentialSession.fetchHandler.bind(credentialSession),
+ );
+ // Delete from both collections (old and new schema) - one or both may exist
+ let rkey = new AtUri(existingSubscription.uri).rkey;
+ await Promise.all([
+ agent.pub.leaflet.graph.subscription
+ .delete({ repo: atp_did, rkey })
+ .catch(() => {}),
+ agent.site.standard.graph.subscription
+ .delete({ repo: atp_did, rkey })
+ .catch(() => {}),
+ ]);
+ }
+
+ await supabaseServerClient
+ .from("publication_subscriptions")
+ .delete()
+ .eq("identity", atp_did)
+ .eq("publication", publication);
+
+ await trackSubscriptionEvent({
+ event: "unsubscribe",
+ method: "atproto",
+ origin: "app",
+ publicationUri: publication,
+ subscriberDid: atp_did,
+ recordUri: existingSubscription.uri,
+ });
+ } catch (e) {
+ console.error(
+ "[deleteAtprotoSubscriptionForDid] failed:",
+ atp_did,
+ publication,
+ e,
+ );
+ }
+}
diff --git a/src/subscriptions/email.ts b/src/subscriptions/email.ts
new file mode 100644
index 00000000..637c5ff2
--- /dev/null
+++ b/src/subscriptions/email.ts
@@ -0,0 +1,286 @@
+import { supabaseServerClient } from "supabase/serverClient";
+import { publishAtprotoSubscriptionForDid } from "src/subscriptions/atproto";
+import { parseActionFromSearchParam } from "app/api/oauth/[route]/afterSignInActions";
+import {
+ getSuppression,
+ deleteSuppression,
+} from "src/utils/postmarkSuppressions";
+import { Ok, Err, type Result } from "src/result";
+import { after } from "next/server";
+import { trackSubscriptionEvent } from "src/subscriptionAnalytics";
+import {
+ sanitizeSubscriptionSource,
+ type SubscriptionSource,
+} from "src/subscriptionSource";
+
+// Upserts the subscriber row for (publication, email) and logs the pair of
+// events every subscription attempt produces: how the subscriber entered — an
+// address that previously unsubscribed re-enters as a resubscribe rather than a
+// fresh request — and what just happened to them. Shared by the code
+// round-trip's `pending` row and the confirmed row every other path writes.
+export async function upsertSubscriber(args: {
+ publicationUri: string;
+ email: string;
+ identityId: string | null;
+ state: "pending" | "confirmed";
+ confirmationCode: string | null;
+ event: "confirmation_sent" | "confirmed";
+}): Promise> {
+ const { data: existing } = await supabaseServerClient
+ .from("publication_email_subscribers")
+ .select("state")
+ .eq("publication", args.publicationUri)
+ .eq("email", args.email)
+ .maybeSingle();
+ const entryEvent =
+ existing?.state === "unsubscribed" ? "resubscribed" : "subscribe_requested";
+
+ const { data: subscriber, error } = await supabaseServerClient
+ .from("publication_email_subscribers")
+ .upsert(
+ {
+ publication: args.publicationUri,
+ email: args.email,
+ identity_id: args.identityId,
+ state: args.state,
+ confirmation_code: args.confirmationCode,
+ unsubscribed_at: null,
+ ...(args.state === "confirmed"
+ ? { confirmed_at: new Date().toISOString() }
+ : {}),
+ },
+ { onConflict: "publication,email" },
+ )
+ .select("id")
+ .single();
+ if (error || !subscriber) {
+ console.error("[upsertSubscriber] upsert failed:", error);
+ return Err("database_error");
+ }
+
+ const { error: eventsError } = await supabaseServerClient
+ .from("publication_email_subscriber_events")
+ .insert(
+ [entryEvent, args.event].map((event_type) => ({
+ subscriber: subscriber.id,
+ publication: args.publicationUri,
+ event_type,
+ })),
+ );
+ if (eventsError) {
+ console.error("[upsertSubscriber] events failed:", eventsError);
+ return Err("database_error");
+ }
+ return Ok(subscriber);
+}
+
+export async function recordEmailSubscription(
+ publicationUri: string,
+ email: string,
+ identityId: string,
+ source?: SubscriptionSource | null,
+): Promise> {
+ const upserted = await upsertSubscriber({
+ publicationUri,
+ email,
+ identityId,
+ state: "confirmed",
+ confirmationCode: null,
+ event: "confirmed",
+ });
+ if (!upserted.ok) return upserted;
+
+ await onEmailSubscriptionConfirmed(publicationUri, email, identityId, source);
+ return Ok(null);
+}
+
+// The tail every path that lands a subscriber row on `confirmed` shares: the
+// analytics event, and the PDS record for a linked atproto account. That
+// record mirrors the same subscription rather than adding a second one, so
+// only the email event is tracked.
+export async function onEmailSubscriptionConfirmed(
+ publicationUri: string,
+ email: string,
+ identityId: string,
+ source?: SubscriptionSource | null,
+): Promise {
+ const { data: confirmedIdentity } = await supabaseServerClient
+ .from("identities")
+ .select("atp_did")
+ .eq("id", identityId)
+ .maybeSingle();
+ after(() =>
+ trackSubscriptionEvent({
+ event: "subscribe",
+ method: "email",
+ origin: "app",
+ publicationUri,
+ subscriberDid: confirmedIdentity?.atp_did,
+ subscriberEmail: email,
+ source,
+ }),
+ );
+ if (confirmedIdentity?.atp_did)
+ await publishAtprotoSubscriptionForDid(
+ confirmedIdentity.atp_did,
+ publicationUri,
+ );
+}
+
+// Flips every still-active subscriber row for this publication to
+// `unsubscribed` and logs the events. This is both "turn off email
+// notifications" and the email half of a full unsubscribe: with the atproto
+// record carrying "subscribed" for linked accounts, an email row's only job is
+// delivery, so muting and unsubscribing the email channel are the same state.
+// Re-enabling goes back through recordEmailSubscription (the resubscribe
+// path), which restores `confirmed`.
+//
+// Rows come from the identity — which may own several historically (different
+// emails tied to the same identity, e.g. a linked address that predates an
+// email change), so one click stops all future mail — plus any `extraRows`
+// resolved outside it, i.e. the row an unsubscribe token points at, which may
+// have no identity at all.
+export async function disableEmailSubscription(
+ publicationUri: string,
+ identity: {
+ id?: string | null;
+ atp_did?: string | null;
+ email?: string | null;
+ } | null,
+ extraRows?: { id: string; state: string }[],
+): Promise> {
+ const rows = [...(extraRows ?? [])];
+ if (identity?.id) {
+ const { data, error } = await supabaseServerClient
+ .from("publication_email_subscribers")
+ .select("id, state")
+ .eq("publication", publicationUri)
+ .eq("identity_id", identity.id);
+ if (error) {
+ console.error("[disableEmailSubscription] read failed:", error);
+ return Err("database_error");
+ }
+ rows.push(...(data ?? []));
+ }
+ const rowIds = [
+ ...new Set(rows.filter((r) => r.state !== "unsubscribed").map((r) => r.id)),
+ ];
+ if (rowIds.length === 0) return Ok(null);
+
+ const nowIso = new Date().toISOString();
+ const [{ error: updateError }, { error: eventError }] = await Promise.all([
+ supabaseServerClient
+ .from("publication_email_subscribers")
+ .update({
+ state: "unsubscribed",
+ unsubscribed_at: nowIso,
+ confirmation_code: null,
+ })
+ .in("id", rowIds),
+ supabaseServerClient.from("publication_email_subscriber_events").insert(
+ rowIds.map((id) => ({
+ subscriber: id,
+ publication: publicationUri,
+ event_type: "unsubscribe_requested",
+ })),
+ ),
+ ]);
+ if (updateError || eventError) {
+ console.error(
+ "[disableEmailSubscription] update/event failed:",
+ updateError ?? eventError,
+ );
+ return Err("database_error");
+ }
+ after(() =>
+ trackSubscriptionEvent({
+ event: "unsubscribe",
+ method: "email",
+ origin: "app",
+ publicationUri,
+ subscriberDid: identity?.atp_did,
+ subscriberEmail: identity?.email,
+ }),
+ );
+ return Ok(null);
+}
+
+export async function checkEmailSubscriptionAllowed(
+ publicationUri: string,
+ email: string,
+): Promise<
+ Result<
+ null,
+ | "newsletter_disabled"
+ | "suppressed_spam_complaint"
+ | "suppression_delete_failed"
+ >
+> {
+ const { data: settings } = await supabaseServerClient
+ .from("publication_newsletter_settings")
+ .select("enabled")
+ .eq("publication", publicationUri)
+ .maybeSingle();
+ if (!settings?.enabled) return Err("newsletter_disabled");
+
+ const suppression = await getSuppression(email);
+ if (suppression?.reason === "SpamComplaint") {
+ return Err("suppressed_spam_complaint");
+ }
+ if (
+ suppression?.reason === "HardBounce" ||
+ suppression?.reason === "ManualSuppression"
+ ) {
+ const deleted = await deleteSuppression(email);
+ if (!deleted) return Err("suppression_delete_failed");
+ }
+ return Ok(null);
+}
+
+// Runs the after-sign-in action (parsed from the `action` search param) for a
+// just-authenticated email identity and returns where to send the browser. When
+// there is no action — or no authenticated identity — the redirect is returned
+// unchanged; only a recognized action does extra work. `subscribe` is currently
+// the only handler. Called from the email-login flow's two mutually-exclusive
+// entry points: the reused-session fast path in app/api/auth/email-login/route.ts
+// and the fresh-confirm path in confirmEmailLogin.
+export async function applyAfterSignInAction(
+ action: string | null,
+ redirect: string,
+ email: string | null,
+ identityId: string | null,
+): Promise {
+ const parsed = parseActionFromSearchParam(action);
+ if (!parsed || !email || !identityId) return redirect;
+
+ switch (parsed.action) {
+ case "subscribe": {
+ const target = new URL(redirect);
+ const allowed = await checkEmailSubscriptionAllowed(
+ parsed.publication,
+ email,
+ );
+ if (!allowed.ok) {
+ target.searchParams.set("subscribe_email_error", allowed.error);
+ return target.toString();
+ }
+ const recorded = await recordEmailSubscription(
+ parsed.publication,
+ email,
+ identityId,
+ sanitizeSubscriptionSource(parsed.source),
+ );
+ if (!recorded.ok) {
+ target.searchParams.set("subscribe_email_error", recorded.error);
+ return target.toString();
+ }
+ // Paid-membership joins carry ?join_tier in the redirect itself, so the
+ // reader resumes payment where they left off rather than detouring here.
+ target.searchParams.set("subscribe_email", email);
+ target.searchParams.set("subscribed_pub", parsed.publication);
+ return target.toString();
+ }
+ default:
+ return redirect;
+ }
+}
diff --git a/src/subscriptions/membership.ts b/src/subscriptions/membership.ts
new file mode 100644
index 00000000..ca11402c
--- /dev/null
+++ b/src/subscriptions/membership.ts
@@ -0,0 +1,67 @@
+import { supabaseServerClient } from "supabase/serverClient";
+import { publishAtprotoSubscriptionForDid } from "src/subscriptions/atproto";
+import {
+ checkEmailSubscriptionAllowed,
+ recordEmailSubscription,
+} from "src/subscriptions/email";
+import type { SubscriptionSource } from "src/subscriptionSource";
+
+// A member is also a subscriber. Called from the inline join flow and from the
+// connect-events webhook's activation paths (requires_action joins only become
+// active there). Best-effort — the membership already billed, so a
+// subscriber-mirroring failure must never fail its caller.
+//
+// The email subscription is skipped when a confirmed one already exists (the
+// sign-in-during-join paths subscribe before payment), so subscriber events and
+// analytics aren't doubled, and when the publication can't mail the address at
+// all. recordEmailSubscription mirrors the join onto the reader's PDS itself;
+// when it didn't run, publish directly so atproto readers still see the
+// subscription (it dedupes internally).
+export async function ensureSubscriberRecordsForMembership(
+ publicationUri: string,
+ identity: { id: string; email: string | null; atp_did: string | null },
+ source?: SubscriptionSource | null,
+ opts?: {
+ // The webhook path also fires on recoveries (past_due → active after a
+ // payment retry), where flipping an `unsubscribed` row back to confirmed
+ // would override a member's deliberate email opt-out. The join flow is the
+ // consent moment, so only it may resubscribe.
+ respectUnsubscribed?: boolean;
+ },
+): Promise {
+ try {
+ let recorded = false;
+ if (identity.email) {
+ const email = identity.email;
+ const [{ data: existing }, allowed] = await Promise.all([
+ supabaseServerClient
+ .from("publication_email_subscribers")
+ .select("state")
+ .eq("publication", publicationUri)
+ .eq("email", email)
+ .maybeSingle(),
+ checkEmailSubscriptionAllowed(publicationUri, email),
+ ]);
+ const skip =
+ !allowed.ok ||
+ existing?.state === "confirmed" ||
+ (opts?.respectUnsubscribed && existing?.state === "unsubscribed");
+ if (!skip)
+ recorded = (
+ await recordEmailSubscription(
+ publicationUri,
+ email,
+ identity.id,
+ source,
+ )
+ ).ok;
+ }
+ if (!recorded && identity.atp_did)
+ await publishAtprotoSubscriptionForDid(identity.atp_did, publicationUri);
+ } catch (e) {
+ console.error(
+ "[ensureSubscriberRecordsForMembership] mirroring failed:",
+ e,
+ );
+ }
+}
diff --git a/src/subscriptions/state.ts b/src/subscriptions/state.ts
new file mode 100644
index 00000000..15d70533
--- /dev/null
+++ b/src/subscriptions/state.ts
@@ -0,0 +1,43 @@
+import {
+ isActiveMembership,
+ type MembershipStatusFields,
+} from "src/membership";
+
+// The one viewer-facing shape for "how am I subscribed to this publication",
+// derived from the three underlying tables (atproto subscription row, email
+// subscriber rows, membership row). Client-safe: used by useViewerSubscription
+// off the identity payload's SUBSCRIPTION_STATE_EMBEDS.
+export type SubscriptionState = {
+ // Any form of subscription, including an active membership. The atproto
+ // record is the root for linked accounts, but an email-only identity's email
+ // row (or membership) counts on its own.
+ subscribed: boolean;
+ atprotoSubscribed: boolean;
+ // A confirmed email row = email delivery on. Muting email flips the row to
+ // `unsubscribed`; there is no separate "muted" state.
+ emailEnabled: boolean;
+};
+
+export function deriveSubscriptionState(
+ publicationUri: string,
+ rows: {
+ subscriptions?: { publication: string }[] | null;
+ emailSubscribers?: { publication: string; state: string }[] | null;
+ memberships?: ({ publication: string } & MembershipStatusFields)[] | null;
+ },
+): SubscriptionState {
+ const atprotoSubscribed = (rows.subscriptions ?? []).some(
+ (s) => s.publication === publicationUri,
+ );
+ const emailEnabled = (rows.emailSubscribers ?? []).some(
+ (s) => s.publication === publicationUri && s.state === "confirmed",
+ );
+ const isMember = (rows.memberships ?? []).some(
+ (m) => m.publication === publicationUri && isActiveMembership(m),
+ );
+ return {
+ subscribed: atprotoSubscribed || emailEnabled || isMember,
+ atprotoSubscribed,
+ emailEnabled,
+ };
+}
diff --git a/src/subscriptions/unsubscribe.ts b/src/subscriptions/unsubscribe.ts
new file mode 100644
index 00000000..b82564c9
--- /dev/null
+++ b/src/subscriptions/unsubscribe.ts
@@ -0,0 +1,108 @@
+import { getStripe } from "stripe/client";
+import { supabaseServerClient } from "supabase/serverClient";
+import { Ok, Err, type Result } from "src/result";
+import { getReaderMembership } from "src/membership.server";
+import { deleteAtprotoSubscriptionForDid } from "src/subscriptions/atproto";
+import { disableEmailSubscription } from "src/subscriptions/email";
+
+export type FullUnsubscribeError =
+ // An active, not-yet-canceled paid membership blocks full unsubscribe: the
+ // reader must cancel (downgrade to the free tier) first, so we never delete
+ // the subscription records out from under a live paid entitlement.
+ | "membership_active"
+ // The membership is already set to cancel at period end; fully unsubscribing
+ // now cancels the Stripe subscription immediately and forfeits the remaining
+ // paid time, so it needs the caller to pass an explicit confirmation.
+ | "membership_pending_cancellation"
+ | "stripe_error"
+ | "database_error";
+
+// A membership Stripe can still bill against. Deliberately broader than
+// isActiveMembership, which reads a membership as inactive the moment
+// current_period_end passes: between period end and the renewal webhook, and
+// throughout a past_due retry, the Stripe subscription is still live, so
+// cancelling it needs the same confirmation an unambiguously active one does.
+function isLiveMembership(m: { status: string | null }): boolean {
+ return (
+ m.status === "active" || m.status === "trialing" || m.status === "past_due"
+ );
+}
+
+// The one full-unsubscribe path: clears the email rows, the atproto record
+// (PDS + row, best-effort on the PDS half), and — for callers that hold both
+// billing flags — an already-canceling Stripe subscription. Used by the
+// authenticated unsubscribe action and the token-link route; `identity` is
+// null for orphaned token rows (no identity_id), which skips the membership
+// and atproto halves. Idempotent: unsubscribing twice is a no-op, not an error.
+export async function fullUnsubscribe(args: {
+ publication: string;
+ identity: { id: string; atp_did: string | null; email: string | null } | null;
+ // Extra rows resolved outside the identity (the token row itself); merged
+ // with the identity's own rows.
+ subscriberRows?: { id: string; state: string }[];
+ cancelPaidImmediately?: boolean;
+ // Only an authenticated caller may set this. Unsubscribe links are followed
+ // by mail-client and link-scanner prefetches and by Postmark's one-click
+ // List-Unsubscribe-Post, none of which represent a human deciding to end a
+ // paid membership, so the token route leaves Stripe untouched entirely.
+ allowStripeCancel?: boolean;
+}): Promise> {
+ const membership = args.identity
+ ? await getReaderMembership(args.publication, args.identity.id)
+ : null;
+
+ if (membership && isLiveMembership(membership)) {
+ if (!membership.cancel_at_period_end) return Err("membership_active");
+ if (!args.cancelPaidImmediately || !args.allowStripeCancel)
+ return Err("membership_pending_cancellation");
+
+ // End the billing before dropping the records, so a Stripe failure leaves
+ // the reader subscribed rather than paying for a publication they no
+ // longer follow.
+ if (membership.stripe_subscription_id && membership.stripe_account_id) {
+ try {
+ await getStripe().subscriptions.cancel(
+ membership.stripe_subscription_id,
+ { stripeAccount: membership.stripe_account_id },
+ );
+ } catch (e) {
+ // A subscription Stripe has already dropped is the outcome we wanted.
+ if ((e as { code?: string })?.code !== "resource_missing") {
+ console.error(
+ `[fullUnsubscribe] failed to cancel subscription ${membership.stripe_subscription_id}:`,
+ e,
+ );
+ return Err("stripe_error");
+ }
+ }
+ const { error } = await supabaseServerClient
+ .from("publication_memberships")
+ .update({ status: "canceled", updated_at: new Date().toISOString() })
+ .eq("id", membership.id);
+ if (error)
+ // The webhook's subscription.deleted event will reconcile the row.
+ console.error("[fullUnsubscribe] membership row update failed:", error);
+ }
+ }
+
+ const disabled = await disableEmailSubscription(
+ args.publication,
+ args.identity,
+ args.subscriberRows,
+ );
+ if (!disabled.ok) return disabled;
+
+ // Idempotent and best-effort; no need to check for a record first.
+ if (args.identity?.atp_did)
+ await deleteAtprotoSubscriptionForDid(
+ args.identity.atp_did,
+ args.publication,
+ );
+
+ // NOTE: Postmark Suppressions API is deliberately NOT called here. Per spec,
+ // in-app unsubscribes only flip local state — this prevents a Pub A
+ // unsubscribe from silently breaking Pub B deliveries on the shared broadcast
+ // stream. Phase 7 handles webhook-driven suppression reconciliation.
+
+ return Ok(null);
+}