From ddb15312ccd64b14f35869bd133d697f91c23ee0 Mon Sep 17 00:00:00 2001
From: Jared Pereira <jared@awarm.space>
Date: Mon, 17 Aug 2026 14:20:39 -0400
Subject: [PATCH] Unify the subscription flows behind src/subscriptions

Email, atproto, and paid membership subscriptions each had their own
notion of "subscribed" spread across the subscribe buttons, the manage
panel, and the unsubscribe paths, so the three surfaces disagreed and
every caller re-derived the state. Collapse them onto
deriveSubscriptionState and a single fullUnsubscribe, replacing
src/emailSubscription.ts with the src/subscriptions module. Muting email
is now a subscriber row in the `unsubscribed` state rather than a new
column.

Cancelling a Stripe subscription is gated behind an allowStripeCancel
flag that only the authenticated action sets. Unsubscribe links are
followed by mail-client prefetches and Postmark's one-click
List-Unsubscribe-Post, so the token route must never end a paid
membership; it stops email and leaves billing intact.

Whether a membership is live is decided by status alone, not by
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 subscription is still live at
Stripe and cancelling it needs the same confirmation an active one does.

The email-notifications toggle only renders when something else keeps
the subscription alive. For an email-only reader, turning it off is the
unsubscribe, and it would drop `subscribed` and unmount the panel
mid-interaction; they get the address as a plain row and the deliberate
Unsubscribe button instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 actions/emailAuth.tsx                         |   2 +-
 actions/publications/joinMembership.ts        |  49 +--
 actions/publications/subscribeEmail.tsx       | 227 ++++----------
 .../publications/subscribeToPublication.ts    | 235 ++------------
 actions/reader/getSubscriptions.ts            |  35 +++
 .../SubscriptionsPageContent.tsx              |   2 +
 .../(writer)/subscriptions/page.tsx           |   5 +-
 .../subscriptions/SubscriptionsContent.tsx    |  18 +-
 app/api/auth/email-login/route.ts             |   2 +-
 .../inngest/functions/send_post_broadcast.ts  |  13 +-
 app/api/oauth/[route]/route.ts                |   6 +-
 .../stripe/connect-events/handlers.ts         |  25 ++
 app/emails/unsubscribe/route.ts               | 195 +++++++-----
 app/test/subscribe/page.tsx                   |   9 +-
 components/Memberships/JoinMembershipFlow.tsx |   4 +-
 components/Subscribe/ManageSubscribe.tsx      | 255 +++++++++++-----
 components/Subscribe/PaidSubscribeButton.tsx  |  30 +-
 components/Subscribe/SubscribeButton.tsx      |  13 +-
 components/Subscribe/viewerSubscription.ts    |  27 +-
 emails/post.tsx                               |  41 ++-
 src/emailSubscription.ts                      | 177 -----------
 src/membership.server.ts                      |   2 +-
 src/mergeIdentity.ts                          |   2 +-
 src/subscriptions/atproto.ts                  | 170 +++++++++++
 src/subscriptions/email.ts                    | 286 ++++++++++++++++++
 src/subscriptions/membership.ts               |  67 ++++
 src/subscriptions/state.ts                    |  43 +++
 src/subscriptions/unsubscribe.ts              | 108 +++++++
 28 files changed, 1218 insertions(+), 830 deletions(-)
 delete mode 100644 src/emailSubscription.ts
 create mode 100644 src/subscriptions/atproto.ts
 create mode 100644 src/subscriptions/email.ts
 create mode 100644 src/subscriptions/membership.ts
 create mode 100644 src/subscriptions/state.ts
 create mode 100644 src/subscriptions/unsubscribe.ts

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<Result<null, UnsubscribeError>> {
   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<void> {
-  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<void> {
-  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<UnsubscribeResult> {
-  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}
         />
       </div>
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<string>();
+  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 <EmptyState title="No subscriptions yet" />;
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, "&amp;"));
               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<UnsubscribeOutcome> {
   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; }
     </style>
   </head>
   <body>
@@ -121,6 +129,17 @@ function htmlPage(title: string, bodyHtml: string): string {
 </html>`;
 }
 
+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 `<a class="manage" href="${esc(manageUrl)}">Manage subscription</a>`;
+}
+
 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",
-        `<p>This unsubscribe link is missing a token. Try opening it directly from the email footer.</p>`,
-      ),
-      { status: 400, headers: { "Content-Type": "text/html; charset=utf-8" } },
+  if (!token)
+    return page(
+      400,
+      "Unsubscribe link invalid",
+      `<p>This unsubscribe link is missing a token. Try opening it directly from the email footer.</p>`,
     );
-  }
-  const result = await unsubscribeByToken(token);
-  if (!result.ok) {
-    return new NextResponse(
-      htmlPage(
-        "Unsubscribe link invalid",
-        `<p>We couldn't find this subscription. It may have already been removed.</p>`,
-      ),
-      {
-        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",
+      `<p>We couldn't find this subscription. It may have already been removed.</p>`,
     );
-  }
-  const emailLine = result.email
-    ? `<p>Future emails to <span class="email">${esc(result.email)}</span> from ${
-        result.publicationName
-          ? esc(result.publicationName)
-          : "this publication"
-      } have been stopped.</p>`
-    : `<p>Your subscription has been removed.</p>`;
-  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 = `<p>Future emails to <span class="email">${esc(outcome.email)}</span> from ${pubName} have been stopped`;
+
+  if (outcome.result === "error")
+    return page(
+      500,
+      "Something went wrong",
+      `<p>We couldn't complete the unsubscribe. Please try again, or manage your subscription directly.</p>
+       ${manageLink(outcome.manageUrl)}`,
+    );
+  if (outcome.result === "membership_active")
+    return page(
+      200,
+      "Emails stopped",
+      `${stopped}.</p>
+       <p>Your paid membership is still active — you can change or cancel it from your subscription settings.</p>
+       ${manageLink(outcome.manageUrl)}`,
+    );
+  return page(
+    200,
+    "Unsubscribed",
+    `${stopped}, and your subscription has been removed.</p>
+     ${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 (
     <Modal
       asChild
+      open={open}
       title={
         <div className="relative mb-2">
           {membershipView?.type === "switch"
@@ -67,8 +92,9 @@ export const ManageSubscription = (props: {
         </div>
       }
       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<boolean | null>(null);
   let toaster = useToaster();
   let router = useRouter();
 
+  const notify = (type: "success" | "error", message: string) =>
+    toaster({ type, content: <div className="font-bold">{message}</div> });
+  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: (
-          <div className="font-bold">
-            {LINK_ERROR_MESSAGES[res.error] ??
-              "We couldn't link that email. Please try again!"}
-          </div>
-        ),
-      });
+      notifyError(res.error, "We couldn't link that email. Please try again!");
       return;
     }
     if (res.value.confirmed) {
-      toaster({
-        content: <div className="font-bold">Email Linked!</div>,
-        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: (
-          <div className="font-bold">
-            {LINK_ERROR_MESSAGES[res.error] ??
-              "We couldn't confirm the code. Please try again!"}
-          </div>
-        ),
-      });
+      notifyError(res.error, "We couldn't confirm the code. Please try again!");
       return;
     }
-    toaster({
-      content: <div className="font-bold">Email Linked!</div>,
-      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) ? (
-          <OAuthErrorMessage error={res.error} />
-        ) : (
-          <div className="font-bold">
-            {UNSUBSCRIBE_ERROR_MESSAGES[res.error] ??
-              "We couldn't unsubscribe you. Please try again!"}
-          </div>
-        ),
-      });
+      notifyError(res.error, "We couldn't unsubscribe you. Please try again!");
       return;
     }
-    toaster({
-      content: <div className="font-bold">Unsubscribed!</div>,
-      type: "success",
-    });
-    refreshIdentityData();
-    router.refresh();
+    notify("success", "Unsubscribed!");
+    mutateMyMembership(props.publicationUri);
+    refreshSubscription();
   };
 
   if (isLoading)
@@ -242,12 +265,22 @@ const ManageSubscriptionContent = (props: {
       </div>
     );
 
-  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 (
     <div className="manageSubPrefs flex flex-col gap-2">
@@ -257,9 +290,18 @@ const ManageSubscriptionContent = (props: {
       {props.newsletterMode && user.email ? (
         <div className={prefClassName}>
           <div className="flex flex-col leading-snug">
-            <p>Linked Email</p>
+            <p>Email Notifications</p>
             <p className="text-tertiary font-normal italic">{user.email}</p>
           </div>
+          {canMuteEmail ? (
+            <Toggle toggle={emailEnabled} onToggle={onToggleEmail}>
+              <span className="sr-only">
+                {emailEnabled
+                  ? "Turn off email notifications"
+                  : "Turn on email notifications"}
+              </span>
+            </Toggle>
+          ) : null}
         </div>
       ) : null}
       {user.handle && (
@@ -338,18 +380,32 @@ const ManageSubscriptionContent = (props: {
       ) : null}
 
       <hr className="border-border-light my-2" />
-      {cancellableMembership ? (
-        <ButtonSecondary
-          fullWidth
-          onClick={() => props.onCancel(cancellableMembership)}
-        >
-          Cancel Membership
-        </ButtonSecondary>
+      {activeMembership && !pendingCancellation ? (
+        <>
+          <ButtonSecondary
+            fullWidth
+            onClick={() => props.onCancel(activeMembership)}
+          >
+            Cancel Membership
+          </ButtonSecondary>
+          <p className="text-tertiary text-sm text-center leading-snug">
+            Cancel your membership first to fully unsubscribe.
+          </p>
+        </>
+      ) : pendingCancellation && confirmForfeit ? (
+        <ForfeitConfirm
+          membership={pendingCancellation}
+          unsubscribing={unsubscribing}
+          onBack={() => setConfirmForfeit(false)}
+          onConfirm={() => onUnsubscribe({ cancelPaidImmediately: true })}
+        />
       ) : (
         <ButtonSecondary
           fullWidth
           disabled={unsubscribing}
-          onClick={onUnsubscribe}
+          onClick={() =>
+            pendingCancellation ? setConfirmForfeit(true) : onUnsubscribe()
+          }
         >
           {unsubscribing ? "Unsubscribing…" : "Unsubscribe"}
         </ButtonSecondary>
@@ -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 (
+    <div className="flex flex-col gap-2">
+      <p className="text-secondary text-sm leading-snug">
+        Your membership{" "}
+        {props.membership.currentPeriodEnd ? (
+          <>
+            lasts until <strong>{endDate}</strong>
+          </>
+        ) : (
+          "is still active"
+        )}
+        . Unsubscribing now cancels it immediately and forfeits the remaining
+        time.
+      </p>
+      <div className="flex gap-2 justify-between">
+        <ButtonSecondary type="button" onClick={props.onBack}>
+          Back
+        </ButtonSecondary>
+        <ButtonPrimary
+          type="button"
+          disabled={props.unsubscribing}
+          onClick={props.onConfirm}
+        >
+          {props.unsubscribing ? <DotLoader /> : "Unsubscribe now"}
+        </ButtonPrimary>
+      </div>
+    </div>
+  );
+};
+
 const MembershipSection = (props: {
   membership: MyMembership;
   onSwitch: (membership: MyMembership) => void;
@@ -394,23 +491,25 @@ const MembershipSection = (props: {
   );
 };
 
-const LINK_ERROR_MESSAGES: Record<string, string> = {
+// Every error code the panel's actions can return; the per-action fallback
+// covers anything not listed.
+const ERROR_MESSAGES: Record<string, string> = {
+  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<string, string> = {
-  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 (
     <>
-      <ManageSubscription
-        publicationUri={props.publicationUri}
-        publicationUrl={props.publicationUrl}
-        newsletterMode={props.newsletterMode}
-        user={user}
-      />
-
+      {user.subscribed ? (
+        <ManageSubscription
+          publicationUri={props.publicationUri}
+          publicationUrl={props.publicationUrl}
+          newsletterMode={props.newsletterMode}
+          user={user}
+        />
+      ) : (
+        <ButtonPrimary
+          compact={props.compact}
+          type="button"
+          className="pubPageSubscribe text-sm!"
+          onClick={() => setOpen(true)}
+        >
+          Subscribe
+        </ButtonPrimary>
+      )}
       {/* 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. */}
       <JoinMembershipModal
         open={open}
         onOpenChange={setOpen}
diff --git a/components/Subscribe/SubscribeButton.tsx b/components/Subscribe/SubscribeButton.tsx
index 9c5e11e2..1fed1e6a 100644
--- a/components/Subscribe/SubscribeButton.tsx
+++ b/components/Subscribe/SubscribeButton.tsx
@@ -154,10 +154,7 @@ export const SubscribeInput = (props: SubscribeProps) => {
     setConfirmOpen(true);
   };
 
-  const showManage = props.newsletterMode
-    ? user.emailSubscribed
-    : user.atprotoSubscribed;
-  const isSubscribed = showManage || locallySubscribed;
+  const isSubscribed = user.subscribed || locallySubscribed;
   const modeMenu = (
     <SubscribeInputModeMenu mode={subscribeMode} onChange={setSubscribeMode} />
   );
@@ -204,7 +201,7 @@ export const SubscribeInput = (props: SubscribeProps) => {
 
           {props.newsletterMode &&
           user.atprotoSubscribed &&
-          !user.emailSubscribed ? (
+          !user.emailEnabled ? (
             <div
               className="text-secondary  w-full text-sm p-2 pt-1.5 mt-1 rounded-md flex flex-col gap-1"
               style={{
@@ -377,10 +374,6 @@ export const SubscribeButton = (props: SubscribeProps) => {
   if (joinable.hasPaidTiers && joinable.tiers)
     return <PaidSubscribeButton {...props} tiers={joinable.tiers} compact />;
 
-  const showManage = props.newsletterMode
-    ? user.emailSubscribed
-    : user.atprotoSubscribed;
-
   const subscribeTrigger = (
     <ButtonPrimary compact className="pubPageSubscribe text-sm!">
       Subscribe
@@ -389,7 +382,7 @@ export const SubscribeButton = (props: SubscribeProps) => {
 
   return (
     <>
-      {showManage || locallySubscribed ? (
+      {user.subscribed || locallySubscribed ? (
         <ManageSubscription
           publicationUri={props.publicationUri}
           publicationUrl={props.publicationUrl}
diff --git a/components/Subscribe/viewerSubscription.ts b/components/Subscribe/viewerSubscription.ts
index 57db231a..6f9e1c8e 100644
--- a/components/Subscribe/viewerSubscription.ts
+++ b/components/Subscribe/viewerSubscription.ts
@@ -1,35 +1,32 @@
 "use client";
 import { useMemo } from "react";
 import { useIdentityData } from "components/IdentityProvider";
+import {
+  deriveSubscriptionState,
+  type SubscriptionState,
+} from "src/subscriptions/state";
 
 export type ViewerUser = {
   loggedIn: boolean;
   email: string | undefined;
   handle: string | undefined;
-  atprotoSubscribed: boolean;
-  emailSubscribed: boolean;
-};
+} & SubscriptionState;
 
 export function useViewerSubscription(publicationUri: string): ViewerUser {
   const { identity } = useIdentityData();
 
   return useMemo(() => {
-    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 `<img src>` verbatim so these must be absolute in outgoing mail.
@@ -935,17 +936,33 @@ export const PostEmail = (props: Partial<PostEmailProps> = {}) => {
                                 }}
                               >
                                 {p.unsubscribeUrl ? (
-                                  <Link
-                                    href={p.unsubscribeUrl}
-                                    style={{
-                                      color: c.tertiary,
-                                      fontSize: 14,
-                                      lineHeight: "20px",
-                                      textDecoration: "underline",
-                                    }}
-                                  >
-                                    Unsubscribe
-                                  </Link>
+                                  [
+                                    p.manageUrl && {
+                                      href: p.manageUrl,
+                                      label: "Manage subscription",
+                                    },
+                                    {
+                                      href: p.unsubscribeUrl,
+                                      label: "Unsubscribe",
+                                    },
+                                  ]
+                                    .filter((l) => !!l)
+                                    .map((l, i) => (
+                                      <Fragment key={l.label}>
+                                        {i > 0 ? " · " : null}
+                                        <Link
+                                          href={l.href}
+                                          style={{
+                                            color: c.tertiary,
+                                            fontSize: 14,
+                                            lineHeight: "20px",
+                                            textDecoration: "underline",
+                                          }}
+                                        >
+                                          {l.label}
+                                        </Link>
+                                      </Fragment>
+                                    ))
                                 ) : (
                                   <span style={{ fontStyle: "italic" }}>
                                     (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<Result<null, "database_error">> {
-  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<string> {
-  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<Result<{ uri: string } | null, OAuthSessionError>> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<Result<{ id: string }, "database_error">> {
+  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<Result<null, "database_error">> {
+  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<void> {
+  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<Result<null, "database_error">> {
+  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<string> {
+  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<void> {
+  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<Result<null, FullUnsubscribeError>> {
+  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);
+}
-- 
2.54.0