diff --git a/pointer/.gitignore b/pointer/.gitignore index b55b0b1..3efb5a7 100644 --- a/pointer/.gitignore +++ b/pointer/.gitignore @@ -45,3 +45,4 @@ yarn-error.log* # typescript *.tsbuildinfo next-env.d.ts +logs/ diff --git a/pointer/app/api/alerts/status/route.ts b/pointer/app/api/alerts/status/route.ts new file mode 100644 index 0000000..d5f67c7 --- /dev/null +++ b/pointer/app/api/alerts/status/route.ts @@ -0,0 +1,48 @@ +/** + * GET /api/alerts/status + * + * Returns the current alarm state for all alerts that apply to the current + * user's active subscription. Includes the alert configuration so the client + * has name/threshold context alongside the status. + */ + +import { headers } from "next/headers"; + +import { auth } from "@/lib/auth"; +import { + listActiveAlarmsForSubscription, + listAlerts, + listApplicableAlertsForSubscription, +} from "@/lib/alerts/provider"; +import { resolveEntitlementSubject } from "@/lib/entitlements/subject"; + +export async function GET(): Promise { + const session = await auth.api.getSession({ headers: await headers() }); + if (!session) return Response.json({ error: "unauthorized" }, { status: 401 }); + + const subject = await resolveEntitlementSubject(session, { + customerType: "user", + }); + if (!subject) return Response.json({ alarms: [], alerts: [] }); + + const [alarms, applicableAlerts, allAlerts] = await Promise.all([ + listActiveAlarmsForSubscription(subject.chargebeeSubscriptionId), + listApplicableAlertsForSubscription(subject.chargebeeSubscriptionId), + listAlerts(), + ]); + + // Merge applicable + all alerts for enrichment (applicable may be empty for + // global alerts that haven't been linked to the subscription yet). + const alertMap = new Map([ + ...allAlerts.map((a) => [a.id, a] as [string, typeof a]), + ...applicableAlerts.map((a) => [a.id, a] as [string, typeof a]), + ]); + + // Enrich each alarm with the alert config for display. + const enriched = alarms.map((alarm) => ({ + ...alarm, + alert: alertMap.get(alarm.alert_id) ?? null, + })); + + return Response.json({ alarms: enriched, total: enriched.length }); +} diff --git a/pointer/app/usage/page.tsx b/pointer/app/usage/page.tsx index 0fe41f1..32a005a 100644 --- a/pointer/app/usage/page.tsx +++ b/pointer/app/usage/page.tsx @@ -2,7 +2,14 @@ import type { Metadata } from "next"; import { headers } from "next/headers"; import { redirect } from "next/navigation"; +import type { Alert, AlertStatus } from "chargebee"; + import { auth } from "@/lib/auth"; +import { + listActiveAlarmsForSubscription, + listAlerts, + listApplicableAlertsForSubscription, +} from "@/lib/alerts/provider"; import { resolveEntitlements } from "@/lib/entitlements/features"; import { getUsageSnapshot } from "@/lib/entitlements/gate"; import { resolveEntitlementSubject } from "@/lib/entitlements/subject"; @@ -33,6 +40,31 @@ type History = | { status: "disabled" } | { status: "unavailable" }; +/** Fetch active alarms for the subscription (best-effort, never throws). */ +async function loadAlarms( + subscriptionId: string, +): Promise> { + try { + const [alarms, applicableAlerts, allAlerts] = await Promise.all([ + listActiveAlarmsForSubscription(subscriptionId), + listApplicableAlertsForSubscription(subscriptionId), + listAlerts(), + ]); + // Merge applicable + all for enrichment — global alerts may not be in + // the "applicable" list until Chargebee links them to the subscription. + const alertMap = new Map([ + ...allAlerts.map((a) => [a.id, a] as [string, typeof a]), + ...applicableAlerts.map((a) => [a.id, a] as [string, typeof a]), + ]); + return alarms.map((alarm) => ({ + ...alarm, + alert: alertMap.get(alarm.alert_id) ?? null, + })); + } catch { + return []; + } +} + /** One aggregate per metered feature — each reduces to a different column. */ async function loadHistory( subscriptionId: string, @@ -82,6 +114,7 @@ export default async function UsagePage({ searchParams }: PageProps<"/usage">) { entitlements.limits, ); const history = await loadHistory(subject.chargebeeSubscriptionId, range); + const activeAlarms = await loadAlarms(subject.chargebeeSubscriptionId); return (
@@ -103,6 +136,45 @@ export default async function UsagePage({ searchParams }: PageProps<"/usage">) {

) : null} + {activeAlarms.length > 0 && ( +
+ {activeAlarms.map((alarm) => { + const name = alarm.alert?.name ?? `Alert ${alarm.alert_id}`; + const threshold = alarm.alert?.threshold; + const thresholdStr = threshold + ? threshold.mode === "percentage" + ? `${threshold.value}%` + : String(threshold.value) + : null; + return ( +
+ + ⚠️ + +
+ {name} + {thresholdStr && ( + + — threshold of {thresholdStr} exceeded + + )} + {alarm.alarm_triggered_at && ( + + since{" "} + {new Date(alarm.alarm_triggered_at * 1000).toLocaleString()} + + )} +
+
+ ); + })} +
+ )} +
diff --git a/pointer/lib/alerts/provider.ts b/pointer/lib/alerts/provider.ts new file mode 100644 index 0000000..df5811c --- /dev/null +++ b/pointer/lib/alerts/provider.ts @@ -0,0 +1,152 @@ +/** + * Thin wrapper around the Chargebee alert and alert_status APIs. + * + * Alert types: + * usage_exceeded — monitors a metered feature; percentage or absolute + * spend_exceeded — monitors overage spend in a currency; absolute only + * credit_balance_dropped — monitors a credit unit balance; absolute only + * + * Scope: + * global (subscription_id omitted) — applies to all subscriptions, optionally + * filtered by plan_price_id via filter_conditions + * subscription-scoped — applies to one subscription only + */ + +import type { Alert, AlertStatus } from "chargebee"; + +import { getRedis } from "@/lib/redis"; +import { chargebeeClient } from "@/plugins/chargebee-plugin"; + +// --------------------------------------------------------------------------- +// Local alarm state — Redis cache keyed by subscription_id. +// Written by the worker when it processes alert_status_changed webhooks so the +// usage page shows alarm banners even before Chargebee's evaluation engine +// has propagated the status to the alertStatus API. +// --------------------------------------------------------------------------- + +const ALARM_KEY_PREFIX = "pointer:alarm:"; +/** TTL = 25 hours, covering the 24-hour billing evaluation window + margin. */ +const ALARM_TTL_SECONDS = 25 * 60 * 60; + +export type LocalAlarmEntry = { + alert_id: string; + subscription_id: string; + alarm_status: "in_alarm" | "ok"; + alarm_triggered_at?: number; + alert_name?: string; + alert_type?: string; + metered_feature_id?: string; +}; + +export async function setLocalAlarm(entry: LocalAlarmEntry): Promise { + const redis = getRedis(); + const key = `${ALARM_KEY_PREFIX}${entry.subscription_id}:${entry.alert_id}`; + if (entry.alarm_status === "ok") { + await redis.del(key); + } else { + await redis.set(key, JSON.stringify(entry), "EX", ALARM_TTL_SECONDS); + } +} + +export async function getLocalAlarms( + subscriptionId: string, +): Promise { + const redis = getRedis(); + const pattern = `${ALARM_KEY_PREFIX}${subscriptionId}:*`; + const keys = await redis.keys(pattern); + if (!keys.length) return []; + const values = await redis.mget(...keys); + return values + .filter(Boolean) + .map((v) => JSON.parse(v!) as LocalAlarmEntry) + .filter((e) => e.alarm_status === "in_alarm"); +} + +export type { Alert, AlertStatus }; + +export async function listAlerts(): Promise { + const result = await chargebeeClient.alert.list({ limit: 100 }); + return result.list.map((entry) => entry.alert); +} + +export async function listApplicableAlertsForSubscription( + subscriptionId: string, +): Promise { + const result = await ( + chargebeeClient.alert as unknown as { + application_alertsForSubscription: ( + id: string, + ) => Promise<{ list: { alert: Alert }[] }>; + } + ).application_alertsForSubscription(subscriptionId); + return result.list.map((entry) => entry.alert); +} + +// alertStatus property exists at runtime but the SDK type file has a typo, +// and the method name uses snake_case with the SDK convention. +const alertStatusResource = (chargebeeClient as unknown as { + alertStatus: { + alert_statusesForSubscription: ( + subscriptionId: string, + params?: { limit?: number; alarm_status?: { is?: string } }, + ) => Promise<{ list: { alert_status: AlertStatus }[] }>; + }; +}).alertStatus; + +export async function listAlertStatusesForSubscription( + subscriptionId: string, +): Promise { + const result = await alertStatusResource.alert_statusesForSubscription( + subscriptionId, + { limit: 100 }, + ); + return result.list.map((entry) => entry.alert_status); +} + +export async function listActiveAlarmsForSubscription( + subscriptionId: string, +): Promise { + // Fetch ALL statuses from Chargebee (not just in_alarm) so we can detect + // when Chargebee has resolved an alarm that's still cached in Redis. + const [cbResult, localAlarms] = await Promise.all([ + alertStatusResource + .alert_statusesForSubscription(subscriptionId, { limit: 100 }) + .catch(() => ({ list: [] as { alert_status: AlertStatus }[] })), + getLocalAlarms(subscriptionId), + ]); + + const allCbStatuses = cbResult.list.map((entry) => entry.alert_status); + const cbAlarms = allCbStatuses.filter((s) => s.alarm_status === "in_alarm"); + + // Build a set of alert_ids Chargebee explicitly marks as "ok". + // These override any stale Redis entry — evict them from Redis asynchronously. + const cbOkAlertIds = new Set( + allCbStatuses + .filter((s) => s.alarm_status === "ok") + .map((s) => s.alert_id), + ); + for (const la of localAlarms) { + if (cbOkAlertIds.has(la.alert_id)) { + // Chargebee says resolved — clear the stale Redis key (fire-and-forget). + setLocalAlarm({ ...la, alarm_status: "ok" }).catch(() => undefined); + } + } + + // Redis fills the gap only for alarms Chargebee hasn't propagated yet + // (not in the Chargebee response at all — neither in_alarm nor ok). + const cbKnownAlertIds = new Set(allCbStatuses.map((s) => s.alert_id)); + const localOnly = localAlarms + .filter((la) => !cbKnownAlertIds.has(la.alert_id)) + .map( + (la) => + ({ + alert_id: la.alert_id, + subscription_id: la.subscription_id, + alarm_status: la.alarm_status, + alarm_triggered_at: la.alarm_triggered_at, + object: "alert_status", + }) as AlertStatus, + ); + + return [...cbAlarms, ...localOnly]; +} diff --git a/pointer/lib/alerts/sync.ts b/pointer/lib/alerts/sync.ts new file mode 100644 index 0000000..5de57b8 --- /dev/null +++ b/pointer/lib/alerts/sync.ts @@ -0,0 +1,98 @@ +/** + * Processes Chargebee `alert_status_changed` webhooks. + * + * When Chargebee evaluates an alert and the status transitions (within_limit + * → in_alarm or back), it fires this event. We emit a domain event so the + * live /admin/flow view and any future notification consumers can react. + */ + +import type { WebhookEvent } from "chargebee"; + +import { emit } from "@/lib/events/emit"; +import { setLocalAlarm } from "@/lib/alerts/provider"; + +const ALERT_STATUS_CHANGED = "alert_status_changed"; + +/** + * Returns true if the event was handled, false if it is not alert-related and + * should continue through the normal webhook pipeline. + */ +export async function processAlertWebhook( + event: WebhookEvent, +): Promise { + if (String(event.event_type) !== ALERT_STATUS_CHANGED) return false; + + const content = event.content as { + alert?: { + id?: string; + name?: string; + type?: string; + metered_feature_id?: string; + threshold?: { mode?: string; value?: number }; + subscription_id?: string | null; + }; + alert_status?: { + alert_id?: string; + subscription_id?: string; + alarm_status?: string; + alarm_triggered_at?: number; + }; + }; + + const alert = content.alert; + const status = content.alert_status; + + if (!alert?.id || !status?.subscription_id) { + console.log("[alert] alert_status_changed received but missing alert.id or subscription_id — skipping", { + eventId: event.id, + alertId: alert?.id, + subscriptionId: status?.subscription_id, + }); + return true; + } + + const isAlarm = status.alarm_status === "in_alarm"; + + console.log(`[alert] ${isAlarm ? "ALARM triggered" : "alarm resolved"}`, { + eventId: event.id, + alertId: alert.id, + alertName: alert.name, + meteredFeature: alert.metered_feature_id, + subscriptionId: status.subscription_id, + alarmStatus: status.alarm_status, + thresholdMode: alert.threshold?.mode, + thresholdValue: alert.threshold?.value, + }); + + await emit( + isAlarm ? "chargebee.alert_triggered" : "chargebee.alert_resolved", + { + alert_id: alert.id, + alert_name: alert.name, + alert_type: alert.type, + metered_feature_id: alert.metered_feature_id, + threshold_mode: alert.threshold?.mode, + threshold_value: alert.threshold?.value, + subscription_id: status.subscription_id, + alarm_status: status.alarm_status, + alarm_triggered_at: status.alarm_triggered_at, + }, + { source: "worker", trace_id: event.id }, + ); + + // Persist alarm state to Redis so the usage page shows banners immediately, + // without waiting for Chargebee's evaluation engine to propagate the status. + await setLocalAlarm({ + alert_id: alert.id, + subscription_id: status.subscription_id, + alarm_status: isAlarm ? "in_alarm" : "ok", + alarm_triggered_at: status.alarm_triggered_at, + alert_name: alert.name, + alert_type: alert.type, + metered_feature_id: alert.metered_feature_id, + }); + + console.log(`[alert] Redis alarm state updated — key pointer:alarm:${status.subscription_id}:${alert.id} → ${isAlarm ? "in_alarm" : "deleted"}`); + + return true; +} diff --git a/pointer/lib/events/types.ts b/pointer/lib/events/types.ts index 0ff8396..0d46143 100644 --- a/pointer/lib/events/types.ts +++ b/pointer/lib/events/types.ts @@ -20,7 +20,9 @@ export type AppEventType = | "app.generate_completed" | "app.usage_threshold" | "app.usage_ingested" - | "app.usage_ingest_failed"; + | "app.usage_ingest_failed" + | "chargebee.alert_triggered" + | "chargebee.alert_resolved"; export interface AppEvent> { event_id: string; diff --git a/pointer/package.json b/pointer/package.json index 2dcb0e3..d8ad6c4 100644 --- a/pointer/package.json +++ b/pointer/package.json @@ -14,6 +14,7 @@ "db:migrate:local": "tsx --env-file=.env.local scripts/migrate-usage.ts && npx --yes @better-auth/cli@latest migrate -y", "bootstrap:chargebee": "tsx --env-file=.env.local scripts/bootstrap-chargebee.ts", "worker:chargebee": "tsx --env-file=.env.local workers/chargebee-webhook-worker.ts", + "worker:chargebee:log": "mkdir -p logs && tsx --env-file=.env.local workers/chargebee-webhook-worker.ts 2>&1 | tee -a logs/worker.log", "reconcile:entitlements": "tsx --env-file=.env.local scripts/reconcile-entitlements.ts" }, "dependencies": { diff --git a/pointer/scripts/bootstrap-chargebee.ts b/pointer/scripts/bootstrap-chargebee.ts index 2c1987f..0318d72 100644 --- a/pointer/scripts/bootstrap-chargebee.ts +++ b/pointer/scripts/bootstrap-chargebee.ts @@ -315,6 +315,39 @@ async function upsertMeteredFeatures() { } } +// --------------------------------------------------------------------------- +// Stage 7: Default Alerts +// +// Creates global usage_exceeded alerts at 80% for each metered feature. +// Idempotent: skips any alert whose name already exists. +// The default threshold is 80% — admins can change it via the /admin/alerts UI. +// --------------------------------------------------------------------------- + +const DEFAULT_ALERT_THRESHOLD = 80; + +async function upsertDefaultAlerts() { + for (const feature of meteredFeatures) { + const name = `${feature.name} — 80% usage`; + const existing = await cb.alert + .list({ limit: 100 }) + .then((r) => r.list.find((e) => e.alert.name === name)?.alert ?? null); + + if (existing) { + log("ok", "alert", existing.id); + continue; + } + + const created = await cb.alert.create({ + type: "usage_exceeded", + name, + description: `Notify when ${feature.name.toLowerCase()} reaches ${DEFAULT_ALERT_THRESHOLD}% of the subscription quota.`, + metered_feature_id: feature.expectedId, + threshold: { mode: "percentage", value: DEFAULT_ALERT_THRESHOLD }, + }); + log("created", "alert", created.alert.id); + } +} + // --------------------------------------------------------------------------- // Main // --------------------------------------------------------------------------- @@ -327,6 +360,7 @@ async function main() { await upsertItemPrices(); await upsertItemEntitlements(); await upsertMeteredFeatures(); + await upsertDefaultAlerts(); console.log("[bootstrap] complete."); } diff --git a/pointer/workers/chargebee-webhook-processor.ts b/pointer/workers/chargebee-webhook-processor.ts index 555d888..857d096 100644 --- a/pointer/workers/chargebee-webhook-processor.ts +++ b/pointer/workers/chargebee-webhook-processor.ts @@ -19,6 +19,7 @@ import { processEntitlementWebhook, runEntitlementSyncJob, } from "@/lib/entitlements/sync"; +import { processAlertWebhook } from "@/lib/alerts/sync"; import { emit } from "@/lib/events/emit"; import { PoisonWebhookError, @@ -198,18 +199,26 @@ export function createChargebeeWebhookMessageProcessor({ return; } - // 3. Dependency pre-check (throws RetryableWebhookError when not ready). - await assertDependencies(event); + // 3a. Handle alert status changes — processed before assertDependencies + // because alert_status_changed events are global and do not require + // the subscription to exist in the local DB mirror yet. + if (await processAlertWebhook(event)) { + // Alert events don't need the rest of the subscription pipeline. + // Fall through to commitVersions + webhook_processed below. + } else { + // 3b. Dependency pre-check (throws RetryableWebhookError when not ready). + await assertDependencies(event); - // 4. Process (plugin DB-sync hooks). - await (await processorPromise).process(event); + // 4. Process (plugin DB-sync hooks). + await (await processorPromise).process(event); - // 5. Verify the plugin's swallowed errors before an entitlement job can - // observe a subscription-created event as successfully mirrored. - await assertProcessed(event); + // 5. Verify the plugin's swallowed errors before an entitlement job can + // observe a subscription-created event as successfully mirrored. + await assertProcessed(event); - // 6. Refresh or queue the resolved entitlement mirror. - await processEntitlementWebhook(event); + // 6. Refresh or queue the resolved entitlement mirror. + await processEntitlementWebhook(event); + } // 7. Commit applied versions, emit, and let the runtime acknowledge. await commitVersions(event);