FEATURE: Add Autumn subscriptions and usage tracking

README.md

@@ -205,7 +205,8 @@ create one D1 database and one private R2 bucket, replace the D1 database ID,
 apply the migration remotely, set the production dashboard origin and
 content domain, and set
 the secrets (`MAINTENANCE_SECRET`, `WORKOS_API_KEY`, `WORKOS_CLIENT_ID`,
-`WORKOS_COOKIE_PASSWORD`, `WORKOS_WEBHOOK_SECRET`):
+`WORKOS_COOKIE_PASSWORD`, `WORKOS_WEBHOOK_SECRET`, `AUTUMN_SECRET_KEY`,
+`AUTUMN_WEBHOOK_SECRET`):
 
 ```bash
 cd apps/web && bun x wrangler secret put MAINTENANCE_SECRET

apps/web/.dev.vars.example

@@ -15,6 +15,13 @@ WORKOS_CLIENT_ID="client_replace"
 WORKOS_COOKIE_PASSWORD="replace-with-at-least-32-characters-of-random-text"
 WORKOS_WEBHOOK_SECRET="replace"
 
+# Autumn billing. Leave AUTUMN_SECRET_KEY empty to run without billing:
+# external billing is disabled; local quotas still apply. `fake:` selects
+# the in-memory fake in development only. The webhook secret is the Svix signing
+# secret of the /api/webhooks/autumn endpoint.
+AUTUMN_SECRET_KEY=""
+AUTUMN_WEBHOOK_SECRET=""
+
 # Local Postgres. `docker compose up -d` at the repo root starts one with
 # these credentials. Wrangler points the HYPERDRIVE binding at it.
 CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_HYPERDRIVE="postgres://adrive:adrive@127.0.0.1:5432/adrive"

apps/web/autumn.config.ts

@@ -0,0 +1,57 @@
+import { feature, item, plan } from 'atmn';
+
+// Plans and features as Autumn config. `bun run billing:push` deploys it;
+// src/lib/server/plans.ts mirrors the limits for the local hard stops and
+// plans.test.ts keeps the two in step. The customer id is the org id.
+
+const GIB = 1024 ** 3;
+
+export const storageBytes = feature({
+	id: 'storage_bytes',
+	name: 'Storage',
+	type: 'metered',
+	consumable: false
+});
+
+export const aiOps = feature({
+	id: 'ai_ops',
+	name: 'AI operations',
+	type: 'metered',
+	consumable: true
+});
+
+export const publicSharing = feature({
+	id: 'public_sharing',
+	name: 'Public sharing',
+	type: 'boolean'
+});
+
+export const free = plan({
+	id: 'free',
+	name: 'Free',
+	autoEnable: true,
+	items: [
+		item({ featureId: storageBytes.id, included: 2 * GIB }),
+		item({
+			featureId: aiOps.id,
+			included: 500,
+			reset: { interval: 'month' }
+		}),
+		item({ featureId: publicSharing.id })
+	]
+});
+
+export const pro = plan({
+	id: 'pro',
+	name: 'Pro',
+	price: { amount: 8, interval: 'month' },
+	items: [
+		item({ featureId: storageBytes.id, included: 100 * GIB }),
+		item({
+			featureId: aiOps.id,
+			included: 10_000,
+			reset: { interval: 'month' }
+		}),
+		item({ featureId: publicSharing.id })
+	]
+});

apps/web/migrations-pg/0010_billing.sql

@@ -0,0 +1,13 @@
+-- migrate:up
+-- Billing counters beside the storage meter. ai_ops_month is this
+-- UTC calendar month's embedded chunk count for the billing page; it
+-- resets on the next increment after ai_ops_month_reset_at. Sync sends
+-- the absolute current total, so a failed acknowledgement is safe to retry.
+ALTER TABLE org_usage
+	ADD COLUMN ai_ops_month integer NOT NULL DEFAULT 0
+		CHECK (ai_ops_month >= 0);
+ALTER TABLE org_usage ADD COLUMN ai_ops_month_reset_at timestamptz;
+
+-- migrate:down
+ALTER TABLE org_usage DROP COLUMN ai_ops_month_reset_at;
+ALTER TABLE org_usage DROP COLUMN ai_ops_month;

apps/web/migrations-pg/0011_usage_delivery.sql

@@ -0,0 +1,23 @@
+-- migrate:up
+-- A periodic absolute snapshot repairs lost enqueues and acknowledgements.
+ALTER TABLE org_usage
+	ADD COLUMN usage_sync_next_run_at timestamptz NOT NULL DEFAULT now();
+
+-- Reserve quota before calling the embedder. Only a successful semantic
+-- commit consumes the reservation and records billable usage. Abandoned
+-- reservations expire with the indexing lease.
+CREATE TABLE ai_usage_reservations (
+	token text PRIMARY KEY,
+	org_id text NOT NULL REFERENCES orgs(id) ON DELETE CASCADE,
+	value integer NOT NULL CHECK (value > 0),
+	expires_at timestamptz NOT NULL
+);
+CREATE INDEX ai_usage_reservations_org ON ai_usage_reservations (org_id, expires_at);
+ALTER TABLE ai_usage_reservations ENABLE ROW LEVEL SECURITY;
+ALTER TABLE ai_usage_reservations FORCE ROW LEVEL SECURITY;
+CREATE POLICY org_isolation ON ai_usage_reservations USING (app_org_visible(org_id));
+
+-- migrate:down
+DROP TABLE ai_usage_reservations;
+ALTER TABLE org_usage
+	DROP COLUMN usage_sync_next_run_at;

apps/web/package.json

@@ -5,6 +5,7 @@
 	"type": "module",
 	"scripts": {
 		"check": "svelte-kit sync && bun run check:types && bun run check:effect && bun run check:svelte",
+		"billing:push": "atmn push",
 		"check:types": "rm -rf .svelte-kit/output .svelte-kit/cloudflare .svelte-kit/cloudflare-tmp && tsc --noEmit",
 		"check:effect": "effect-tsgo diagnostics --project tsconfig.json --format pretty",
 		"check:svelte": "svelte-check --tsconfig ./tsconfig.json --tsgo",
@@ -19,6 +20,7 @@
 		"@effect/sql-pg": "4.0.0-beta.102",
 		"@modelcontextprotocol/server": "2.0.0",
 		"agents": "^0.20.1",
+		"autumn-js": "1.3.0",
 		"effect": "4.0.0-beta.102",
 		"markdown-it": "^14.3.0",
 		"mime": "^4.1.0",
@@ -39,6 +41,7 @@
 		"@types/pg": "^8.23.1",
 		"@typescript/native": "npm:typescript@^7.0.2",
 		"@workos-inc/node": "10.13.0",
+		"atmn": "1.1.24",
 		"concurrently": "^10.0.4",
 		"svelte": "5.56.8",
 		"svelte-check": "4.7.4",

apps/web/src/env.d.ts

@@ -16,6 +16,8 @@ declare global {
 		CF_ACCOUNT_ID?: string;
 		CF_API_TOKEN?: string;
 		CF_ZONE_ID?: string;
+		AUTUMN_SECRET_KEY?: string;
+		AUTUMN_WEBHOOK_SECRET?: string;
 	}
 
 	namespace Cloudflare {
@@ -31,6 +33,8 @@ declare global {
 			CF_ACCOUNT_ID?: string;
 			CF_API_TOKEN?: string;
 			CF_ZONE_ID?: string;
+			AUTUMN_SECRET_KEY?: string;
+			AUTUMN_WEBHOOK_SECRET?: string;
 		}
 	}
 }

apps/web/src/lib/dashboard/api.ts

@@ -10,6 +10,8 @@ import type {
 import {
 	parseApiKeyCreateResponse,
 	parseApiKeyListResponse,
+	parseBillingLink,
+	parseBillingSummary,
 	parseFileContentLink,
 	parseFileDetailResponse,
 	parseFileListResponse,
@@ -90,6 +92,54 @@ export const changeOrgSlug = async (token: string, slug: string) => {
 	return json(parseOrgSettings, response);
 };
 
+// Billing requests need a deadline through body decoding as well as fetch.
+// Clear our timer when the request settles or its caller cancels it.
+const withBillingTimeout = async <A>(
+	signal: AbortSignal | undefined,
+	run: (signal: AbortSignal) => Promise<A>
+) => {
+	const timeout = new AbortController();
+	const timer = setTimeout(() => {
+		timeout.abort(new Error('Billing request timed out. Please try again.'));
+	}, 15_000);
+	try {
+		return await run(
+			signal ? AbortSignal.any([signal, timeout.signal]) : timeout.signal
+		);
+	} catch (cause) {
+		if (timeout.signal.aborted) throw timeout.signal.reason;
+		throw cause;
+	} finally {
+		clearTimeout(timer);
+	}
+};
+
+export const getBilling = (token: string, signal?: AbortSignal) =>
+	withBillingTimeout(signal, async (requestSignal) => {
+		const response = await request('/api/billing', token, {
+			signal: requestSignal
+		});
+		return json(parseBillingSummary, response);
+	});
+
+export const startCheckout = (token: string, signal?: AbortSignal) =>
+	withBillingTimeout(signal, async (requestSignal) => {
+		const response = await request('/api/billing/checkout', token, {
+			method: 'POST',
+			signal: requestSignal
+		});
+		return (await json(parseBillingLink, response)).url;
+	});
+
+export const openBillingPortal = (token: string, signal?: AbortSignal) =>
+	withBillingTimeout(signal, async (requestSignal) => {
+		const response = await request('/api/billing/portal', token, {
+			method: 'POST',
+			signal: requestSignal
+		});
+		return (await json(parseBillingLink, response)).url;
+	});
+
 export const listApiKeys = async (token: string, signal?: AbortSignal) => {
 	const response = await request('/api/auth/keys', token, { signal });
 	return (await json(parseApiKeyListResponse, response)).keys;

apps/web/src/lib/dashboard/parse.ts

@@ -8,6 +8,8 @@
 // Struct decoders, they require every declared field but ignore extra keys.
 import type {
 	ApiKey,
+	BillingLinkResponse,
+	BillingSummary,
 	DashboardFile,
 	FileContentLinkResponse,
 	FileDetailResponse,
@@ -271,6 +273,31 @@ export const parseOrgSettings = (value: unknown): OrgSettings => {
 	};
 };
 
+export const parseBillingSummary = (value: unknown): BillingSummary => {
+	const record = requireRecord(value, 'billing');
+	const storage = requireRecord(record.storage, 'billing.storage');
+	const aiOps = requireRecord(record.aiOps, 'billing.aiOps');
+	return {
+		plan: text(record.plan, 'billing.plan'),
+		planName: text(record.planName, 'billing.planName'),
+		billingEnabled: flag(record.billingEnabled, 'billing.billingEnabled'),
+		canManageBilling: flag(record.canManageBilling, 'billing.canManageBilling'),
+		storage: {
+			used: integer(storage.used, 'billing.storage.used'),
+			limit: integer(storage.limit, 'billing.storage.limit')
+		},
+		aiOps: {
+			used: integer(aiOps.used, 'billing.aiOps.used'),
+			limit: integer(aiOps.limit, 'billing.aiOps.limit')
+		}
+	};
+};
+
+export const parseBillingLink = (value: unknown): BillingLinkResponse => {
+	const record = requireRecord(value, 'billing');
+	return { url: maybeString(record.url, 'billing.url') };
+};
+
 export const parseApiKeyCreateResponse = (
 	value: unknown
 ): ApiKeyCreateResponse => {

apps/web/src/lib/server/billing-webhook.ts

@@ -0,0 +1,53 @@
+import type { PgClient } from '@effect/sql-pg';
+import { Effect } from 'effect';
+import { StorageError } from './errors';
+import type { AutumnClientShape } from './services/autumn';
+
+// plan_changes contains only affected plans, not the complete subscription
+// state. A verified customer-level event asks us to reconcile from Autumn.
+export type BillingWebhookEvent =
+	| { readonly kind: 'reconcile'; readonly orgId: string }
+	| { readonly kind: 'ignored'; readonly type: string };
+
+const isRecord = (value: unknown): value is Record<string, unknown> =>
+	typeof value === 'object' && value !== null && !Array.isArray(value);
+
+const text = (value: unknown) => (typeof value === 'string' ? value : '');
+
+export const decodeBillingWebhook = (payload: unknown): BillingWebhookEvent => {
+	if (!isRecord(payload)) return { kind: 'ignored', type: '' };
+	const type = text(payload.type);
+	const data = isRecord(payload.data) ? payload.data : null;
+	const orgId = data ? text(data.customer_id) : '';
+	if (type !== 'billing.updated' || !orgId || !data || text(data.entity_id)) {
+		return { kind: 'ignored', type };
+	}
+	return { kind: 'reconcile', orgId };
+};
+
+// Lock before the provider read, so concurrent/redelivered signals cannot
+// finish an older read after a newer one and restore a stale local plan.
+// Unknown customers are acknowledged without creating an Autumn customer.
+export const reconcileOrgPlan = (
+	sql: PgClient.PgClient,
+	autumn: AutumnClientShape,
+	orgId: string
+) =>
+	sql
+		.withTransaction(
+			Effect.gen(function* () {
+				const rows = yield* sql<{ id: string }>`
+		SELECT id FROM orgs WHERE id = ${orgId} FOR UPDATE`;
+				if (rows.length === 0) return null;
+				const plan = yield* autumn.getPlan({ customerId: orgId });
+				yield* sql`UPDATE orgs SET plan = ${plan} WHERE id = ${orgId}`;
+				return plan;
+			})
+		)
+		.pipe(
+			Effect.mapError((cause) =>
+				cause instanceof StorageError
+					? cause
+					: new StorageError({ operation: 'reconcile org plan', cause })
+			)
+		);

apps/web/src/lib/server/config.ts

@@ -26,6 +26,14 @@ export interface CloudflareZoneConfig {
 	readonly zoneId: string;
 }
 
+// Autumn billing. secretKey is null when AUTUMN_SECRET_KEY is unset, in
+// which case every gate fails open; a `fake:` key picks the in-memory
+// fake only in development. See services/autumn.ts.
+export interface AutumnConfig {
+	readonly secretKey: string | null;
+	readonly webhookSecret: string;
+}
+
 export interface AppConfigShape {
 	readonly dashboardOrigin: string;
 	// Tenant content is served from `<slug>.<contentDomain>` over the
@@ -41,6 +49,7 @@ export interface AppConfigShape {
 	readonly cloudflareZone: CloudflareZoneConfig | null;
 	// WorkOS user ids allowed on /admin (ADMIN_USER_IDS, comma-separated).
 	readonly adminUserIds: ReadonlySet<string>;
+	readonly autumn: AutumnConfig;
 	readonly semanticSearch: 'off' | 'auto' | 'required';
 	readonly embeddingModel: '@cf/baai/bge-small-en-v1.5';
 	readonly embeddingPooling: 'cls';
@@ -123,6 +132,19 @@ const adminUserIdsFromEnv = (env: Env) =>
 			.map((id) => id.trim())
 			.filter((id) => id.length > 0)
 	);
+const autumnFromEnv = (env: Env): AutumnConfig => {
+	const secretKey = optionalString(env.AUTUMN_SECRET_KEY).trim() || null;
+	const webhookSecret = optionalString(env.AUTUMN_WEBHOOK_SECRET).trim();
+	if (secretKey?.startsWith('fake:') && !dev) {
+		throw new Error('Fake Autumn billing is only available in development');
+	}
+	if (secretKey && !secretKey.startsWith('fake:') && !webhookSecret) {
+		throw new Error(
+			'AUTUMN_WEBHOOK_SECRET is required alongside AUTUMN_SECRET_KEY'
+		);
+	}
+	return { secretKey, webhookSecret };
+};
 
 export const configFromEnv = (env: Env) => {
 	const origins = normalizeOrigins({
@@ -160,6 +182,7 @@ export const configFromEnv = (env: Env) => {
 		urlScanner: urlScannerFromEnv(env),
 		cloudflareZone: cloudflareZoneFromEnv(env),
 		adminUserIds: adminUserIdsFromEnv(env),
+		autumn: autumnFromEnv(env),
 		semanticSearch,
 		embeddingModel: '@cf/baai/bge-small-en-v1.5',
 		embeddingPooling: 'cls',

apps/web/src/lib/server/edge.ts

@@ -20,6 +20,9 @@ import { requestLayer } from './layer';
 import { PgSql } from './pg';
 import type { AppConfig } from './config';
 import type { Auth } from './services/auth';
+import type { AutumnClient } from './services/autumn';
+import type { Billing } from './services/billing';
+import type { BillingGates } from './services/billing-gates';
 import type { AuthGuardStore } from './services/bindings';
 import type { Blobs } from './services/blobs';
 import type { CurrentOrg, CurrentUser } from './services/current-org';
@@ -64,7 +67,10 @@ export type AppServices =
 	| GrantSecrets
 	| JobQueue
 	| Org
-	| WorkOSClient;
+	| WorkOSClient
+	| AutumnClient
+	| BillingGates
+	| Billing;
 
 export const isAppError = (failure: unknown): failure is AppError =>
 	failure instanceof InvalidRequest ||