feat: ship connected publishing and Cloud billing

services/cloud-api/migrations/0003_publish_releases.sql

@@ -0,0 +1,81 @@
+PRAGMA foreign_keys = ON;
+
+CREATE TABLE sites (
+  id TEXT PRIMARY KEY,
+  entry_id TEXT NOT NULL,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  slug TEXT NOT NULL UNIQUE,
+  title TEXT NOT NULL,
+  current_release_id TEXT,
+  created_at INTEGER NOT NULL,
+  updated_at INTEGER NOT NULL
+) STRICT;
+
+CREATE UNIQUE INDEX sites_owner_entry ON sites(user_id, entry_id);
+CREATE INDEX sites_public_slug ON sites(slug);
+
+CREATE TABLE publish_sessions (
+  id TEXT PRIMARY KEY,
+  site_id TEXT,
+  entry_id TEXT NOT NULL,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  title TEXT NOT NULL,
+  manifest_key TEXT NOT NULL,
+  manifest_hash TEXT NOT NULL,
+  expires_at INTEGER NOT NULL,
+  created_at INTEGER NOT NULL
+) STRICT;
+
+CREATE INDEX publish_sessions_expiry ON publish_sessions(expires_at);
+
+CREATE TABLE publish_session_objects (
+  session_id TEXT NOT NULL REFERENCES publish_sessions(id) ON DELETE CASCADE,
+  user_id TEXT NOT NULL,
+  content_hash TEXT NOT NULL,
+  PRIMARY KEY (session_id, content_hash)
+) STRICT;
+
+CREATE INDEX publish_session_objects_lookup
+  ON publish_session_objects(user_id, content_hash);
+
+CREATE TABLE releases (
+  id TEXT PRIMARY KEY,
+  site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
+  manifest_key TEXT NOT NULL,
+  manifest_hash TEXT NOT NULL,
+  page_count INTEGER NOT NULL,
+  asset_count INTEGER NOT NULL,
+  published_at INTEGER NOT NULL
+) STRICT;
+
+CREATE TABLE stored_objects (
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  content_hash TEXT NOT NULL,
+  object_key TEXT NOT NULL,
+  kind TEXT NOT NULL CHECK (kind IN ('page', 'asset')),
+  content_type TEXT NOT NULL,
+  byte_size INTEGER NOT NULL,
+  created_at INTEGER NOT NULL,
+  PRIMARY KEY (user_id, content_hash)
+) STRICT;
+
+CREATE TABLE release_objects (
+  release_id TEXT NOT NULL REFERENCES releases(id) ON DELETE CASCADE,
+  user_id TEXT NOT NULL,
+  content_hash TEXT NOT NULL,
+  PRIMARY KEY (release_id, content_hash),
+  FOREIGN KEY (user_id, content_hash)
+    REFERENCES stored_objects(user_id, content_hash) ON DELETE CASCADE
+) STRICT;
+
+CREATE INDEX release_objects_lookup ON release_objects(user_id, content_hash);
+
+CREATE TABLE legacy_publish_objects (
+  object_key TEXT PRIMARY KEY
+) STRICT;
+
+INSERT OR IGNORE INTO legacy_publish_objects (object_key)
+SELECT object_key FROM shares;
+
+DROP TRIGGER IF EXISTS shares_free_limit_before_insert;
+DROP TABLE IF EXISTS shares;

services/cloud-api/migrations/0004_auth_retention.sql

@@ -0,0 +1 @@
+CREATE INDEX sessions_expiry ON sessions(expires_at);

services/cloud-api/migrations/0005_otp_rate_events.sql

@@ -0,0 +1,37 @@
+CREATE TABLE otp_rate_events (
+  id TEXT PRIMARY KEY,
+  email_fingerprint TEXT NOT NULL,
+  request_fingerprint TEXT NOT NULL,
+  created_at INTEGER NOT NULL
+) STRICT;
+
+CREATE INDEX otp_rate_events_email_created
+  ON otp_rate_events(email_fingerprint, created_at DESC);
+
+CREATE INDEX otp_rate_events_request_created
+  ON otp_rate_events(request_fingerprint, created_at DESC);
+
+CREATE INDEX otp_rate_events_created
+  ON otp_rate_events(created_at);
+
+CREATE TRIGGER otp_rate_events_email_limit
+BEFORE INSERT ON otp_rate_events
+WHEN (
+  SELECT COUNT(*) FROM otp_rate_events
+  WHERE email_fingerprint = NEW.email_fingerprint
+    AND created_at >= NEW.created_at - 3600000
+) >= 5
+BEGIN
+  SELECT RAISE(ABORT, 'otp_email_rate_limit');
+END;
+
+CREATE TRIGGER otp_rate_events_request_limit
+BEFORE INSERT ON otp_rate_events
+WHEN (
+  SELECT COUNT(*) FROM otp_rate_events
+  WHERE request_fingerprint = NEW.request_fingerprint
+    AND created_at >= NEW.created_at - 3600000
+) >= 20
+BEGIN
+  SELECT RAISE(ABORT, 'otp_request_rate_limit');
+END;

services/cloud-api/migrations/0006_dodo_billing.sql

@@ -0,0 +1,33 @@
+PRAGMA foreign_keys = ON;
+
+ALTER TABLE entitlements ADD COLUMN provider TEXT;
+ALTER TABLE entitlements ADD COLUMN provider_customer_id TEXT;
+ALTER TABLE entitlements ADD COLUMN provider_subscription_id TEXT;
+ALTER TABLE entitlements ADD COLUMN billing_interval TEXT;
+ALTER TABLE entitlements ADD COLUMN provider_status TEXT;
+ALTER TABLE entitlements ADD COLUMN provider_updated_at INTEGER NOT NULL DEFAULT 0;
+
+CREATE UNIQUE INDEX entitlements_provider_customer
+  ON entitlements(provider, provider_customer_id)
+  WHERE provider_customer_id IS NOT NULL;
+
+CREATE UNIQUE INDEX entitlements_provider_subscription
+  ON entitlements(provider, provider_subscription_id)
+  WHERE provider_subscription_id IS NOT NULL;
+
+CREATE TABLE billing_handoffs (
+  token_hash TEXT PRIMARY KEY,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  expires_at INTEGER NOT NULL,
+  used_at INTEGER,
+  created_at INTEGER NOT NULL
+) STRICT;
+
+CREATE INDEX billing_handoffs_expiry ON billing_handoffs(expires_at);
+
+CREATE TABLE billing_webhooks (
+  id TEXT PRIMARY KEY,
+  event_type TEXT NOT NULL,
+  processed_at INTEGER NOT NULL
+) STRICT;
+

services/cloud-api/src/auth.ts

@@ -29,9 +29,10 @@ export async function authenticatedUser(
      JOIN users ON users.id = sessions.user_id
      LEFT JOIN entitlements ON entitlements.user_id = users.id
        AND entitlements.status IN ('active', 'trialing')
+       AND (entitlements.current_period_end IS NULL OR entitlements.current_period_end > ?)
      WHERE sessions.token_hash = ? AND sessions.expires_at > ?`,
   )
-    .bind(await sha256(token), Date.now())
+    .bind(Date.now(), await sha256(token), Date.now())
     .first<SessionRow>();
   if (!row) throw new AuthenticationError();
 

services/cloud-api/src/billing.ts

@@ -0,0 +1,226 @@
+import { authenticatedUser } from "./auth";
+import { randomToken, sha256 } from "./crypto";
+import {
+  createCheckoutSession,
+  createPortalSession,
+  DodoApiError,
+  DodoWebhookError,
+  type DodoSubscription,
+  verifyDodoWebhook,
+} from "./dodo";
+import { json, readJson } from "./http";
+import type { AuthenticatedUser, BillingInterval, Env } from "./types";
+
+const HANDOFF_TTL_MS = 15 * 60 * 1000;
+const SUBSCRIPTION_EVENTS = new Set([
+  "subscription.active",
+  "subscription.updated",
+  "subscription.on_hold",
+  "subscription.renewed",
+  "subscription.plan_changed",
+  "subscription.cancelled",
+  "subscription.failed",
+  "subscription.expired",
+]);
+
+interface HandoffRow {
+  user_id: string;
+  email: string;
+}
+
+interface EntitlementLookup {
+  user_id: string;
+}
+
+export class BillingError extends Error {
+  constructor(
+    readonly status: number,
+    readonly code: string,
+    message: string,
+  ) {
+    super(message);
+  }
+}
+
+export async function createBillingHandoff(request: Request, env: Env): Promise<Response> {
+  const user = await authenticatedUser(request, env);
+  const token = randomToken();
+  const now = Date.now();
+  const expiresAt = now + HANDOFF_TTL_MS;
+  await env.DB.prepare(
+    `INSERT INTO billing_handoffs (token_hash, user_id, expires_at, created_at)
+     VALUES (?, ?, ?, ?)`,
+  ).bind(await sha256(token), user.id, expiresAt, now).run();
+  const pricingUrl = new URL("/pricing", env.PUBLIC_SITE_ORIGIN);
+  pricingUrl.searchParams.set("billing_token", token);
+  return json({ url: pricingUrl.toString(), expiresAt });
+}
+
+export async function beginCheckout(request: Request, env: Env): Promise<Response> {
+  const input = checkoutInput(await readJson(request, 4_096));
+  const tokenHash = await sha256(input.token);
+  const usedAt = Date.now();
+  const handoff = await env.DB.prepare(
+    `UPDATE billing_handoffs SET used_at = ?
+     WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?
+     RETURNING user_id, (SELECT email FROM users WHERE id = user_id) AS email`,
+  ).bind(usedAt, tokenHash, usedAt).first<HandoffRow>();
+  if (!handoff) {
+    throw new BillingError(401, "billing_link_expired", "Open pricing from Markd again to continue.");
+  }
+
+  const user: AuthenticatedUser = { id: handoff.user_id, email: handoff.email, plan: "free" };
+  try {
+    return json({ checkoutUrl: await createCheckoutSession(env, user, input.interval) });
+  } catch (cause) {
+    await env.DB.prepare(
+      "UPDATE billing_handoffs SET used_at = NULL WHERE token_hash = ? AND used_at = ?",
+    ).bind(tokenHash, usedAt).run();
+    if (cause instanceof DodoApiError) throw new BillingError(502, "checkout_unavailable", cause.message);
+    throw cause;
+  }
+}
+
+export async function billingPortal(request: Request, env: Env): Promise<Response> {
+  const user = await authenticatedUser(request, env);
+  const row = await env.DB.prepare(
+    "SELECT provider_customer_id FROM entitlements WHERE user_id = ? AND provider = 'dodo'",
+  ).bind(user.id).first<{ provider_customer_id: string | null }>();
+  if (!row?.provider_customer_id) {
+    throw new BillingError(404, "billing_customer_missing", "No billing account exists for this user.");
+  }
+  try {
+    return json({ url: await createPortalSession(env, row.provider_customer_id) });
+  } catch (cause) {
+    if (cause instanceof DodoApiError) throw new BillingError(502, "portal_unavailable", cause.message);
+    throw cause;
+  }
+}
+
+export async function dodoWebhook(request: Request, env: Env): Promise<Response> {
+  const rawBody = await request.text();
+  let verified;
+  try {
+    verified = await verifyDodoWebhook(rawBody, request.headers, env.DODO_PAYMENTS_WEBHOOK_KEY);
+  } catch (cause) {
+    if (cause instanceof DodoWebhookError) throw new BillingError(400, "invalid_webhook", cause.message);
+    throw cause;
+  }
+  if (verified.payload.business_id !== env.DODO_BUSINESS_ID) {
+    throw new BillingError(400, "invalid_webhook", "The webhook business does not match Markd.");
+  }
+
+  const duplicate = await env.DB.prepare("SELECT 1 FROM billing_webhooks WHERE id = ?")
+    .bind(verified.id).first();
+  if (duplicate) return json({ received: true, duplicate: true });
+
+  if (SUBSCRIPTION_EVENTS.has(verified.payload.type)) {
+    await syncSubscription(env, verified.payload.type, verified.payload.timestamp, verified.payload.data);
+  }
+  await env.DB.prepare(
+    "INSERT OR IGNORE INTO billing_webhooks (id, event_type, processed_at) VALUES (?, ?, ?)",
+  ).bind(verified.id, verified.payload.type, Date.now()).run();
+  return json({ received: true });
+}
+
+export async function cleanupBillingRecords(env: Env): Promise<void> {
+  const now = Date.now();
+  await env.DB.batch([
+    env.DB.prepare("DELETE FROM billing_handoffs WHERE expires_at <= ?").bind(now),
+    env.DB.prepare("DELETE FROM billing_webhooks WHERE processed_at <= ?")
+      .bind(now - 90 * 24 * 60 * 60 * 1000),
+  ]);
+}
+
+async function syncSubscription(
+  env: Env,
+  eventType: string,
+  eventTimestamp: string,
+  subscription: DodoSubscription,
+): Promise<void> {
+  if (!subscription.subscription_id || !subscription.product_id) return;
+  const interval = productInterval(env, subscription.product_id);
+  if (!interval) return;
+  const userId = await subscriptionUserId(env, subscription);
+  if (!userId) throw new BillingError(409, "billing_user_missing", "The subscription has no Markd user.");
+
+  const providerUpdatedAt = Date.parse(eventTimestamp);
+  const periodEnd = subscription.next_billing_date
+    ? Date.parse(subscription.next_billing_date)
+    : null;
+  const status = entitlementStatus(eventType, subscription.status, periodEnd);
+  await env.DB.prepare(
+    `INSERT INTO entitlements (
+       user_id, plan, status, current_period_end, provider, provider_customer_id,
+       provider_subscription_id, billing_interval, provider_status, provider_updated_at
+     ) VALUES (?, 'cloud', ?, ?, 'dodo', ?, ?, ?, ?, ?)
+     ON CONFLICT(user_id) DO UPDATE SET
+       plan = 'cloud', status = excluded.status,
+       current_period_end = excluded.current_period_end, provider = 'dodo',
+       provider_customer_id = excluded.provider_customer_id,
+       provider_subscription_id = excluded.provider_subscription_id,
+       billing_interval = excluded.billing_interval,
+       provider_status = excluded.provider_status,
+       provider_updated_at = excluded.provider_updated_at
+     WHERE excluded.provider_updated_at >= entitlements.provider_updated_at`,
+  ).bind(
+    userId,
+    status,
+    Number.isFinite(periodEnd) ? periodEnd : null,
+    subscription.customer?.customer_id ?? null,
+    subscription.subscription_id,
+    interval,
+    subscription.status ?? eventType.replace("subscription.", ""),
+    Number.isFinite(providerUpdatedAt) ? providerUpdatedAt : Date.now(),
+  ).run();
+}
+
+async function subscriptionUserId(env: Env, subscription: DodoSubscription): Promise<string | null> {
+  const metadataUser = subscription.metadata?.markd_user_id;
+  if (metadataUser) {
+    const user = await env.DB.prepare("SELECT id FROM users WHERE id = ?")
+      .bind(metadataUser).first<{ id: string }>();
+    if (user) return user.id;
+  }
+  const customerId = subscription.customer?.customer_id;
+  if (customerId) {
+    const entitlement = await env.DB.prepare(
+      "SELECT user_id FROM entitlements WHERE provider = 'dodo' AND provider_customer_id = ?",
+    ).bind(customerId).first<EntitlementLookup>();
+    if (entitlement) return entitlement.user_id;
+  }
+  const email = subscription.customer?.email?.trim().toLowerCase();
+  if (!email) return null;
+  const user = await env.DB.prepare("SELECT id AS user_id FROM users WHERE email = ?")
+    .bind(email).first<EntitlementLookup>();
+  return user?.user_id ?? null;
+}
+
+function checkoutInput(value: unknown): { token: string; interval: BillingInterval } {
+  if (!value || typeof value !== "object") throw new BillingError(400, "invalid_checkout", "Checkout details are required.");
+  const input = value as Record<string, unknown>;
+  if (typeof input.token !== "string" || !/^[A-Za-z0-9_-]{43}$/.test(input.token)) {
+    throw new BillingError(400, "invalid_checkout", "The billing link is invalid.");
+  }
+  if (input.interval !== "monthly" && input.interval !== "yearly") {
+    throw new BillingError(400, "invalid_checkout", "Choose monthly or yearly billing.");
+  }
+  return { token: input.token, interval: input.interval };
+}
+
+function productInterval(env: Env, productId: string): BillingInterval | null {
+  if (productId === env.DODO_MONTHLY_PRODUCT_ID) return "monthly";
+  if (productId === env.DODO_YEARLY_PRODUCT_ID) return "yearly";
+  return null;
+}
+
+export function entitlementStatus(
+  eventType: string,
+  providerStatus: string | undefined,
+  periodEnd: number | null,
+): "active" | "past_due" | "canceled" {
+  if (providerStatus === "active" || eventType === "subscription.active" || eventType === "subscription.renewed") return "active";
+  if (providerStatus === "on_hold" || eventType === "subscription.on_hold") return "past_due";
+  if (providerStatus === "cancelled" && periodEnd && periodEnd > Date.now()) return "active";
+  return "canceled";
+}