paddle-subscription-sync

将 Paddle 订阅状态同步到数据库

何时使用此技能

当你需要在本地保存一份 Paddle 订阅和客户状态副本时，通常会用到此技能：例如，根据订阅状态控制功能访问权限、无需每次都调用 Paddle API 即可呈现账户页面，以及报告收入。此技能定义了表结构、要订阅的事件、upsert 模式、如何将应用的用户映射到 Paddle 客户、状态语义，以及如何处理已安排的变更（暂停、在周期结束时生效的取消）。

此技能假设你已接收经过验证的 webhook。如果还没有，请先完成 webhooks——该技能介绍了路由处理程序和签名验证。此处的代码会放在该技能创建的事件处理函数中。

前提条件

• 可正常运行且经过签名验证的 webhook 端点（webhooks）。
• 一个数据库（示例使用 Supabase，但任何支持 UPSERT 的 SQL 存储都可以）。
• 通知目标已订阅以下事件：
  • customer.created、customer.updated
  • subscription.created、subscription.updated、subscription.canceled
  • 可选：transaction.completed（针对一次性购买）、subscription.activated（试用转为正式订阅时）
• 对于回填和对账任务：需要拥有 customer.read 和 subscription.read 权限的 API 密钥。webhook 处理程序本身不会调用 API。

为什么要镜像状态？

每次需要知道“这个用户是否订阅了 Pro 方案？”时，你都可以调用 Paddle API——但与查询自己的数据库相比，这样做速度较慢。

镜像意味着：webhook 是更新数据库的事实来源；应用读取自己的数据库；API 则留给写入操作（取消、更新、退款），以便立即获得响应。

架构

最精简可用的 SQL 架构：

-- Maps a Paddle customer to your app user (email is the fallback bridge).
CREATE TABLE customers (
  customer_id TEXT PRIMARY KEY,        -- "ctm_01h..."
  email TEXT NOT NULL,
  last_event_at TIMESTAMPTZ,           -- occurredAt of the newest event applied
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX customers_email_idx ON customers(email);

CREATE TABLE subscriptions (
  subscription_id TEXT PRIMARY KEY,    -- "sub_01h..."
  customer_id TEXT NOT NULL REFERENCES customers(customer_id),
  user_id TEXT,                        -- your app's user ID, from checkout customData
  subscription_status TEXT NOT NULL,   -- see "Status semantics" below
  price_id TEXT NOT NULL,
  product_id TEXT NOT NULL,
  scheduled_change_action TEXT,        -- 'cancel' | 'pause' | 'resume'; null if nothing is scheduled
  scheduled_change_effective_at TIMESTAMPTZ,
  paddle_created_at TIMESTAMPTZ NOT NULL, -- Paddle's created_at, for ordering
  last_event_at TIMESTAMPTZ,           -- occurredAt of the newest event applied
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX subscriptions_customer_id_idx ON subscriptions(customer_id);
CREATE INDEX subscriptions_user_id_idx ON subscriptions(user_id);
CREATE INDEX subscriptions_status_idx ON subscriptions(subscription_status);

此架构有意保持精简。根据需要添加列（例如 currency_code、next_billed_at、discount_id，以及用于多商品订阅的 JSONB 类型 items）。Paddle 数据要丰富得多；只需镜像 UI 实际查询的内容。

Paddle 不保证事件按顺序送达：较新的事件之后，可能才收到重试或重放的事件。因此，写入通过两个函数进行：在一条原子语句中插入一行，或仅在传入事件的时间晚于 last_event_at 时更新该行：

CREATE OR REPLACE FUNCTION upsert_subscription(
  p_subscription_id TEXT, p_customer_id TEXT, p_user_id TEXT, p_status TEXT,
  p_price_id TEXT, p_product_id TEXT, p_scheduled_change_action TEXT,
  p_scheduled_change_effective_at TIMESTAMPTZ, p_paddle_created_at TIMESTAMPTZ,
  p_last_event_at TIMESTAMPTZ
) RETURNS void LANGUAGE sql AS $$
  INSERT INTO subscriptions AS s (
    subscription_id, customer_id, user_id, subscription_status, price_id, product_id,
    scheduled_change_action, scheduled_change_effective_at, paddle_created_at, last_event_at
  ) VALUES (
    p_subscription_id, p_customer_id, p_user_id, p_status, p_price_id, p_product_id,
    p_scheduled_change_action, p_scheduled_change_effective_at, p_paddle_created_at, p_last_event_at
  )
  ON CONFLICT (subscription_id) DO UPDATE SET
    customer_id = EXCLUDED.customer_id,
    user_id = COALESCE(EXCLUDED.user_id, s.user_id),
    subscription_status = EXCLUDED.subscription_status,
    price_id = EXCLUDED.price_id,
    product_id = EXCLUDED.product_id,
    scheduled_change_action = EXCLUDED.scheduled_change_action,
    scheduled_change_effective_at = EXCLUDED.scheduled_change_effective_at,
    paddle_created_at = EXCLUDED.paddle_created_at,
    last_event_at = EXCLUDED.last_event_at,
    updated_at = NOW()
  WHERE s.last_event_at IS NULL OR s.last_event_at < EXCLUDED.last_event_at;
$$;

CREATE OR REPLACE FUNCTION upsert_customer(
  p_customer_id TEXT, p_email TEXT, p_last_event_at TIMESTAMPTZ
) RETURNS void LANGUAGE sql AS $$
  INSERT INTO customers AS c (customer_id, email, last_event_at)
  VALUES (p_customer_id, p_email, p_last_event_at)
  ON CONFLICT (customer_id) DO UPDATE SET
    email = EXCLUDED.email,
    last_event_at = EXCLUDED.last_event_at,
    updated_at = NOW()
  WHERE c.last_event_at IS NULL OR c.last_event_at < EXCLUDED.last_event_at;
$$;

这些函数以调用者的权限运行，因此行级安全策略仍然适用；请使用服务角色客户端调用它们。IS NULL 检查可让第一条写入落到在添加 last_event_at 之前就已存在的行上。在 Supabase 之外，直接运行相同的 INSERT … ON CONFLICT … DO UPDATE … WHERE 语句即可。

Supabase 客户端

示例使用两个服务端 Supabase 辅助函数，与 Paddle Next.js 入门套件一致。如果你的应用已有对应实现，请使用它们。@/utils/supabase/server 中的 createClient 会以已登录用户的身份读取数据（匿名密钥、用户会话 Cookie）：

// utils/supabase/server.ts
import { createServerClient } from "@supabase/ssr";
import { cookies } from "next/headers";

export async function createClient() {
  const cookieStore = await cookies();
  return createServerClient(
    process.env.NEXT_PUBLIC_SUPABASE_URL!,
    process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!,
    {
      cookies: {
        getAll: () => cookieStore.getAll(),
        setAll(cookiesToSet) {
          try {
            cookiesToSet.forEach(({ name, value, options }) =>
              cookieStore.set(name, value, options),
            );
          } catch {
            // Called from a Server Component; middleware refreshes the session instead.
          }
        },
      },
    },
  );
}

// utils/supabase/server-internal.ts — service role, no user session. Server-only.
import { createServerClient } from "@supabase/ssr";

export async function createClient() {
  // No cookies: with the signed-in user's session attached, requests would run
  // as that user and row-level security (RLS) would still apply.
  return createServerClient(
    process.env.NEXT_PUBLIC_SUPABASE_URL!,
    process.env.SUPABASE_SERVICE_ROLE_KEY!,
    { cookies: { getAll: () => [], setAll: () => {} } },
  );
}

将其导入为 import { createClient as createAdminClient } from "@/utils/supabase/server-internal"，并且仅在服务端使用：在 webhook 处理程序中，或在确认用户身份后的 Server Action 中使用。它会有意跳过 Cookie，因此永远不会获取已登录用户的会话。如果你编写自己的封装，请勿将其命名为 createServerClient，以免与 @supabase/ssr 导入项重名。

将应用用户映射到 Paddle 客户

最可靠的关联方式是通过结账传递你自己的用户 ID：

• 使用 customData: { userId: user.id } 打开结账（参见 checkout-web）。
• Paddle 会将结账的自定义数据复制到交易中，以及由此创建的订阅中。subscription.created 事件会携带 data.customData.userId。
• 以下处理程序会将其存储为 subscriptions.user_id。

对于未携带该自定义数据创建的订阅，请将电子邮件作为备用匹配方式：

• 你应用的用户表中有电子邮件。
• Paddle 的客户记录中有电子邮件。
• 用户进入账单页面时，按电子邮件匹配。

customers.email 镜像表让你无需调用 API，就能根据已通过身份验证的用户查找 Paddle 的 customer_id：

// utils/paddle/get-customer-id.ts
import { createClient } from "@/utils/supabase/server";

export async function getCustomerId(): Promise<string> {
  const supabase = await createClient();
  const { data: userData } = await supabase.auth.getUser();
  const email = userData?.user?.email;
  if (!email) return "";

  const { data } = await supabase
    .from("customers")
    .select("customer_id")
    .eq("email", email)
    .single();

  return data?.customer_id ?? "";
}

如果用户先在你的应用中注册，然后才结账（此时还没有 Paddle 客户），getCustomerId 会返回 ''。首次结账后，customer.created webhook 会触发，处理程序会插入或更新该行，之后的调用就能找到该用户。

处理订阅事件

在 webhooks 中的路由处理程序内，将事件路由到带类型的处理程序：

// utils/paddle/process-webhook.ts
import {
  EventName,
  type Customer,
  type CustomerCreatedEvent,
  type CustomerUpdatedEvent,
  type EventEntity,
  type Subscription,
  type SubscriptionCanceledEvent,
  type SubscriptionCreatedEvent,
  type SubscriptionUpdatedEvent,
} from "@paddle/paddle-node-sdk";
import { createClient as createAdminClient } from "@/utils/supabase/server-internal";

type SubscriptionEvent =
  | SubscriptionCreatedEvent
  | SubscriptionUpdatedEvent
  | SubscriptionCanceledEvent;
type CustomerEvent = CustomerCreatedEvent | CustomerUpdatedEvent;

export async function processEvent(event: EventEntity) {
  switch (event.eventType) {
    case EventName.SubscriptionCreated:
    case EventName.SubscriptionUpdated:
    case EventName.SubscriptionCanceled:
      return saveSubscription(event.data, event.occurredAt);
    case EventName.CustomerCreated:
    case EventName.CustomerUpdated:
      return saveCustomer(event.data, event.occurredAt);
  }
}

// eventAt is the event's occurredAt (or the entity's updatedAt when reconciling).
// The SQL function ignores the write if the row already has a newer one.
export async function saveSubscription(
  sub: SubscriptionEvent["data"] | Subscription,
  eventAt: string,
) {
  const admin = await createAdminClient();
  const userId = sub.customData?.userId;

  const { error } = await admin.rpc("upsert_subscription", {
    p_subscription_id: sub.id,
    p_customer_id: sub.customerId,
    p_user_id: typeof userId === "string" ? userId : null,
    p_status: sub.status,
    p_price_id: sub.items[0]?.price?.id ?? "",
    p_product_id: sub.items[0]?.price?.productId ?? "",
    p_scheduled_change_action: sub.scheduledChange?.action ?? null,
    p_scheduled_change_effective_at: sub.scheduledChange?.effectiveAt ?? null,
    p_paddle_created_at: sub.createdAt,
    p_last_event_at: eventAt,
  });
  if (error) throw error;
}

export async function saveCustomer(customer: CustomerEvent["data"] | Customer, eventAt: string) {
  const admin = await createAdminClient();
  const { error } = await admin.rpc("upsert_customer", {
    p_customer_id: customer.id,
    p_email: customer.email,
    p_last_event_at: eventAt,
  });
  if (error) throw error;
}

几点说明：

• webhook 处理程序必须使用带有服务角色密钥的 Supabase 客户端（createAdminClient），而不是 anon 密钥——webhook 不会以已登录用户的身份运行，RLS 策略会阻止 anon 用户写入。将服务角色客户端放在单独的文件（server-internal.ts）中，以确保它不会被导入客户端代码。
• 顺序：较旧事件在重试或重放后，可能会在较新事件之后到达。由于这些函数仅在 occurredAt 晚于 last_event_at 时才写入数据，延迟到达和重复投递都不会造成任何更改。对于发送电子邮件等非状态写入副作用，请继续根据 event.eventId 去重（参见 webhooks）。
• items 数组可能包含多个明细项，用于混合订阅。此示例仅存储第一项；对于多明细项订阅，请将整个数组存储为 JSONB。

状态含义

你的访问控制逻辑应读取 subscription.status 字段。其值如下：

状态	含义	用户应有权访问吗？
active	正在付费，当前有效。	是
trialing	处于试用期（免费或付费）。	是
past_due	付款逾期。自动扣款：Paddle 会重试（催收）。手动收款（发票）：不会重试。	是，显示横幅提示
paused	订阅已暂停；不会计费，也不提供服务。	否（通常如此）
canceled	取消已生效。	否

canceled 是终止状态——表示订阅确实已经结束。如果用户取消订阅，但当前计费周期尚未结束，状态不会改变（仍为 active、trialing 或 past_due），且 scheduled_change_action 为 cancel。只有在 scheduled_change_effective_at 到达时，状态才会变为 canceled。

一种合理的访问控制逻辑：

function hasActiveSubscription(sub: Subscription | null): boolean {
  if (!sub) return false;
  return ["active", "trialing", "past_due"].includes(sub.subscription_status);
}

处于 past_due 状态期间，应继续允许用户访问，但显示横幅，提示用户更新付款方式（如果是手动收款，则提示支付发票）。

处理计划变更

当用户在计费周期中途取消订阅，或安排暂停时，Paddle 会发送一个 subscription.updated 事件，其中包含：

{
  "data": {
    "id": "sub_01h...",
    "status": "active",
    "scheduled_change": {
      "action": "cancel",
      "effective_at": "2026-05-12T00:00:00Z"
    }
  }
}

状态不会改变（此处为 active；也可能是 trialing 或 past_due）——用户会继续使用服务，直到 effective_at。随后，Paddle 会发送另一个事件，将状态改为 canceled（或 paused），并清除 scheduled_change。带有恢复日期的暂停订阅会有一个 resume 计划变更。

将 scheduled_change.action 和 scheduled_change.effective_at 存储为 scheduled_change_action 和 scheduled_change_effective_at。在界面中，根据操作决定显示的消息，例如显示“您的订阅将于 5 月 12 日取消”，而不是“订阅”。以下是一种常见模式：

function getSubscriptionUiState(sub: Subscription | null) {
  if (!sub) return "no-subscription";
  switch (sub.scheduled_change_action) {
    case "cancel":
      return "cancel-scheduled";
    case "pause":
      return "pause-scheduled";
    case "resume":
      return "resume-scheduled";
    default:
      return sub.subscription_status; // 'active' | 'trialing' | 'past_due' | 'paused' | 'canceled'
  }
}

查询镜像状态

完成镜像后，无需调用 API，即可从数据库读取所有内容：

// utils/db/get-user-subscription.ts
"use server";

import { createClient } from "@/utils/supabase/server";
import { getCustomerId } from "@/utils/paddle/get-customer-id";

export async function getUserSubscription() {
  const supabase = await createClient();
  const { data: userData } = await supabase.auth.getUser();
  const userId = userData?.user?.id;
  if (!userId) return null;

  // Match the user ID from checkout customData, or fall back to the email-matched customer.
  const customerId = await getCustomerId();
  const match = customerId
    ? `user_id.eq.${userId},customer_id.eq.${customerId}`
    : `user_id.eq.${userId}`;

  const { data } = await supabase
    .from("subscriptions")
    .select("*")
    .or(match)
    .order("paddle_created_at", { ascending: false })
    .limit(1)
    .maybeSingle();

  return data;
}

这会返回 Paddle 的 created_at 最晚的订阅。不要按自己的 created_at 排序：它表示行写入的时间，回填和乱序事件会打乱顺序。如果用户可以持有多个订阅，请按 subscription_status 筛选（例如 .in("subscription_status", ["active", "trialing", "past_due"])），或返回所有订阅。

仅在进行变更时使用 Paddle API：取消、暂停、恢复、升级、退款。读取应从数据库进行。

初始回填

如果要在现有应用中添加订阅同步（且 Paddle 中已有数据），需要先进行回填，再依赖镜像数据。

Paddle MCP 每个服务会公开三个工具（search、execute、report_missing_tool）。工作流：先调用 search，确认准确的方法名称和参数结构，然后调用 execute，传入一个异步函数来调用 client.<resource>.<operation>(...)。方法路径采用 camelCase（client.clientTokens.create、client.pricingPreview.preview）。请求体参数和响应字段采用 snake_case（tax_category、product_id、unit_price、currency_code）。列表方法返回 { pagination: { hasMore, estimatedTotal }, <resource>: [...] }，而不是 data。数组的键是资源的 API 路径段，因此当方法路径采用 camelCase 时，数组键会采用 kebab-case：(await client.products.list()).products、(await client.clientTokens.list())["client-tokens"]。当 pagination.hasMore 为 true 时，使用 { after: "<last_id>" } 翻页，而不是调用 .next()。未分页的列表（例如 client.currencies.list()）会返回普通数组。单实体方法会返回该实体。在一次 execute 中串联多步骤工作流；变量不会在不同调用之间持久保留。硬性上限：每次 execute 最多 50 次 API 调用、超时 30 秒，代码最多 32,000 个字符。

如果你可以使用 Paddle MCP 服务，请在一次 execute 调用中显式翻页：

async (client) => {
  const customers = [];
  let after;
  do {
    const page = await client.customers.list({
      after,
      per_page: 200,
      status: ["active", "archived"],
    });
    customers.push(...page.customers);
    after = page.pagination.hasMore ? page.customers.at(-1).id : undefined;
  } while (after);
  return customers;
};

默认情况下，客户列表只会返回 active 客户；请包含 archived，这样较早的订阅也能找到对应的客户行。使用相同方式翻阅 client.subscriptions.list()（page.subscriptions）；默认会返回所有状态的订阅。

这里的 50 次调用上限很重要。 如果设置 per_page: 200，每次 execute 调用最多可处理约 10,000 个客户。如果用户的账户数据量可以在一次 execute 中处理完，这样就没问题。否则，请拆分为多次 execute 调用——每次都将上次看到的最后一个 ID 作为起始 after 游标传入——或者使用下方的 SDK 作业，它没有此类上限。

否则（或要处理更大的回填），请使用 SDK。通过相同的 saveCustomer/saveSubscription 辅助函数写入，可以保留顺序保护机制：每个实体的 updatedAt 都作为事件时间，因此作业不会覆盖较新的 webhook 写入。请在 Next.js 应用中运行，以便复用这些辅助函数，例如使用 cron 触发的 Route Handler：

// app/api/cron/reconcile-paddle/route.ts
import { getPaddleInstance } from "@/utils/paddle/get-paddle-instance";
import { saveCustomer, saveSubscription } from "@/utils/paddle/process-webhook";

export async function GET(request: Request) {
  const secret = process.env.CRON_SECRET;
  if (!secret || request.headers.get("authorization") !== `Bearer ${secret}`) {
    return new Response("Unauthorized", { status: 401 });
  }

  const paddle = getPaddleInstance();
  for await (const customer of paddle.customers.list({ status: ["active", "archived"] })) {
    await saveCustomer(customer, customer.updatedAt);
  }
  for await (const sub of paddle.subscriptions.list()) {
    await saveSubscription(sub, sub.updatedAt);
  }
  return Response.json({ ok: true });
}

请在将访问控制切换为从数据库读取之前运行一次。之后按计划持续运行（例如由平台的 cron 每晚运行），作为对账作业修复任何因 webhook 丢失而导致的过期数据。对于大型账户，请拆分工作，以免超出平台函数的运行时限。

常见陷阱

• 在 webhook 处理程序中使用 Supabase anon 密钥。 RLS 策略会阻止写入。请在服务器内部客户端中使用 service role 密钥（createAdminClient）。
• 在结账前通过 customer_id 映射用户。 新应用用户在完成结账前没有 Paddle 客户。请在结账 customData 中传入用户 ID，并在必要时回退到按电子邮件匹配。
• 把 canceled 当作“刚刚点击了取消”。 这是终止状态。“用户点击了取消”对应的事件是 subscription.updated，其 scheduled_change.action: 'cancel' 非空，除非你允许用户立即取消。在状态变为 canceled 之前，不要撤销访问权限。
• 从过时的行读取状态。 如果处理程序使用队列（处理耗时操作时应该如此），数据库行会在队列工作线程运行后才更新。Webhook 收到后到行更新之间的延迟通常不到一秒；但如果你需要在路由处理程序本身内实现写后读，请同步执行 upsert。
• 只存储 subscription.items 中的一个项目。 如果你销售混合套餐（基础订阅 + 附加项），请存储整个数组。否则 UI 只会显示第一个项目。
• 切换为从数据库读取前没有回填数据。 现有客户会显示为“无订阅”，直到收到下一个 webhook——而这可能发生在他们取消订阅时。务必先回填，再按计划持续对账。
• 忘记处理 customer.updated。 如果客户在 Paddle 中更改了电子邮件，在同步新电子邮件之前，你将无法通过 getCustomerId 查找他们。从一开始就订阅 customer.updated。
• 事件顺序错乱。 Paddle 不保证事件的投递顺序。subscription.updated 可能先于 subscription.created 到达，重试或重放的旧事件也可能在新事件之后到达。简单的 UPSERT 会让旧事件覆盖较新的状态。请通过上面的 last_event_at 防护逻辑写入，并使用 event.eventId 对副作用进行去重。

验证集成

1. 通过结账流程创建一个沙盒订阅，并在 customData 中传入用户 ID。
2. 检查 customers 和 subscriptions 表——在 checkout.completed 发生后的一秒内，两张表都应出现新行，并且订阅行中的 user_id 已设置。
3. 通过客户门户取消订阅。
4. 确认 subscriptions 行保留其状态（active，或试用期间的 trialing），且 scheduled_change_action: 'cancel' 和 scheduled_change_effective_at 已设置。
5. 沙盒中没有测试时钟，因此不要等待周期结束。可以使用 effective_from: "immediately" 取消沙盒订阅（参见 subscription-cancel），也可以运行模拟器的 subscription_cancellation 场景（参见 sandbox-testing）；该场景默认立即生效。模拟器只会发送 webhooks，因此请填写沙盒订阅的 ID，以便在负载中获取其真实 ID，并使用 Usage type 为 Simulation 或 Both 的目标。确认该行的状态变为 canceled，且计划变更列已清空。
6. 以相同方式运行 subscription_pause 场景，并确认 UI 能正确处理 paused。

相关文档

• 通过 webhooks 配置访问权限
• 取消订阅
• 暂停订阅
• 参考实现：paddle-nextjs-starter-kit — 参见 src/utils/paddle/process-webhook.ts 和 src/utils/paddle/get-customer-id.ts。