konnosDashboard
polar-meter-push.ts271 lines · main
1import { and, asc, eq } from 'drizzle-orm';
2
3import { getDb } from '../db/client.js';
4import { projects, subscriptions, usageEvents, type UsageEvent, type UsageMetric } from '../db/schema.js';
5import { env } from '../env.js';
6import { log } from '../lib/logger.js';
7
8/**
9 * Polar metering push — drains pending usage_events rows into Polar's
10 * meter API. Each metric maps to one Polar meter id:
11 *
12 * invocations → BRIVEN_POLAR_METER_INVOCATIONS_ID
13 * storage_bytes → BRIVEN_POLAR_METER_STORAGE_ID
14 * connection_seconds → BRIVEN_POLAR_METER_CONNECTION_ID
15 * auth_mau → BRIVEN_POLAR_METER_AUTH_MAU_ID
16 * auth_sso_connections → BRIVEN_POLAR_METER_AUTH_SSO_CONNECTIONS_ID
17 * auth_sso_signins → BRIVEN_POLAR_METER_AUTH_SSO_SIGNINS_ID
18 *
19 * Without the meter ids configured the worker marks rows as 'skipped'
20 * so they don't pile up forever (and a follow-up env config flip can
21 * re-enable them by resetting status='pending').
22 *
23 * Push status transitions (see usageEvents.polarPushStatus):
24 * pending → pushed (HTTP 2xx from Polar)
25 * pending → skipped (no meter id configured for the metric)
26 * pending → pending (transient failure, retried next tick)
27 */
28
29/** Batch size — keep round-trips small so a single Polar outage doesn't lock up the cron. */
30const BATCH_SIZE = 50;
31/** Cron interval — once per minute. Each tick drains up to BATCH_SIZE rows. */
32const INTERVAL_MS = 60_000;
33
34/**
35 * Polar customer id required for the meter payload. Resolved through
36 * the project's org's subscription row — subscriptions are org-level
37 * (CLAUDE.md §3.3) and Polar tracks billing per customer at the same
38 * grain. Returns null when the org has no subscription yet, in which
39 * case the worker marks the row pending → skipped with reason
40 * 'no_customer' so it doesn't pile up forever; the next checkout
41 * stamps the customer id and a follow-up cron can re-enable skipped
42 * rows by resetting status='pending'.
43 *
44 * In-process cache with a 5-minute TTL — every pending row hits this
45 * during a drain and we don't want to hammer the meta-DB. Subscription
46 * upserts on the Polar webhook path invalidate the entry so a fresh
47 * customer id takes effect on the next tick, not after the TTL.
48 */
49interface CustomerCacheEntry {
50 customerId: string | null;
51 expiresAt: number;
52}
53const CUSTOMER_CACHE_TTL_MS = 5 * 60_000;
54const customerCache = new Map<string, CustomerCacheEntry>();
55
56export function invalidatePolarCustomerCache(projectId: string): void {
57 customerCache.delete(projectId);
58}
59
60async function polarCustomerForProject(projectId: string): Promise<string | null> {
61 const now = Date.now();
62 const hit = customerCache.get(projectId);
63 if (hit && hit.expiresAt > now) {
64 return hit.customerId;
65 }
66 const db = getDb();
67 const [row] = await db
68 .select({ polarCustomerId: subscriptions.polarCustomerId })
69 .from(projects)
70 .innerJoin(subscriptions, eq(subscriptions.orgId, projects.orgId))
71 .where(eq(projects.id, projectId))
72 .limit(1);
73 const customerId = row?.polarCustomerId ?? null;
74 customerCache.set(projectId, { customerId, expiresAt: now + CUSTOMER_CACHE_TTL_MS });
75 return customerId;
76}
77
78function meterIdFor(metric: UsageMetric): string | null {
79 switch (metric) {
80 case 'invocations':
81 return env.BRIVEN_POLAR_METER_INVOCATIONS_ID ?? null;
82 case 'storage_bytes':
83 return env.BRIVEN_POLAR_METER_STORAGE_ID ?? null;
84 case 'connection_seconds':
85 return env.BRIVEN_POLAR_METER_CONNECTION_ID ?? null;
86 case 'auth_mau':
87 return env.BRIVEN_POLAR_METER_AUTH_MAU_ID ?? null;
88 case 'auth_sso_connections':
89 return env.BRIVEN_POLAR_METER_AUTH_SSO_CONNECTIONS_ID ?? null;
90 case 'auth_sso_signins':
91 return env.BRIVEN_POLAR_METER_AUTH_SSO_SIGNINS_ID ?? null;
92 default:
93 return null;
94 }
95}
96
97/**
98 * Push one row. Returns the new status — caller writes it back.
99 *
100 * Without an access token + meter id + customer id the row is marked
101 * 'skipped'. With all three present we POST to Polar's Meters API.
102 * Network or 5xx → 'pending' (retried on the next tick); 4xx → 'skipped'
103 * (operator must intervene — a 400 from Polar is durable, not transient).
104 *
105 * Operator-visible logs:
106 * polar_push_pushed — 2xx response, row → 'pushed'
107 * polar_push_skipped_* — see reason field; row → 'skipped'
108 * polar_push_retry — transient failure; row stays 'pending'
109 */
110async function pushOne(row: UsageEvent): Promise<'pushed' | 'skipped' | 'pending'> {
111 if (!env.BRIVEN_POLAR_ACCESS_TOKEN) {
112 log.info('polar_push_skipped_no_token', {
113 eventId: row.id,
114 projectId: row.projectId,
115 metric: row.metric,
116 });
117 return 'skipped';
118 }
119 const meterId = meterIdFor(row.metric);
120 if (!meterId) {
121 log.info('polar_push_skipped_no_meter', {
122 eventId: row.id,
123 projectId: row.projectId,
124 metric: row.metric,
125 });
126 return 'skipped';
127 }
128 const customerId = await polarCustomerForProject(row.projectId);
129 if (!customerId) {
130 log.info('polar_push_skipped_no_customer', {
131 eventId: row.id,
132 projectId: row.projectId,
133 metric: row.metric,
134 });
135 return 'skipped';
136 }
137
138 // Polar Meters API — POST {base}/v1/meters/{meter_id}/events
139 // { customer_id, value: number, timestamp: ISO8601 }
140 const value = Number.parseFloat(row.value);
141 if (!Number.isFinite(value)) {
142 log.warn('polar_push_skipped_bad_value', {
143 eventId: row.id,
144 projectId: row.projectId,
145 metric: row.metric,
146 raw: row.value,
147 });
148 return 'skipped';
149 }
150
151 let res: Response;
152 try {
153 res = await fetch(`${env.BRIVEN_POLAR_API_BASE}/v1/meters/${meterId}/events`, {
154 method: 'POST',
155 headers: {
156 authorization: `Bearer ${env.BRIVEN_POLAR_ACCESS_TOKEN}`,
157 'content-type': 'application/json',
158 },
159 body: JSON.stringify({
160 customer_id: customerId,
161 value,
162 timestamp: row.periodStart.toISOString(),
163 }),
164 });
165 } catch (err) {
166 log.warn('polar_push_retry', {
167 eventId: row.id,
168 projectId: row.projectId,
169 metric: row.metric,
170 reason: 'network',
171 message: err instanceof Error ? err.message : String(err),
172 });
173 return 'pending';
174 }
175
176 if (res.ok) {
177 log.info('polar_push_pushed', {
178 eventId: row.id,
179 projectId: row.projectId,
180 metric: row.metric,
181 meterId,
182 value,
183 });
184 return 'pushed';
185 }
186
187 // 5xx is transient (Polar restarted, rate-limited, etc.) — leave the
188 // row pending and let the next tick retry. 4xx is durable (bad meter
189 // id, customer mismatch, validation) — mark skipped so we don't loop
190 // forever; an operator can re-enable after fixing the underlying issue.
191 const transient = res.status >= 500 || res.status === 429;
192 const body = await res.text().catch(() => '');
193 log.warn(transient ? 'polar_push_retry' : 'polar_push_skipped_4xx', {
194 eventId: row.id,
195 projectId: row.projectId,
196 metric: row.metric,
197 status: res.status,
198 body: body.slice(0, 256),
199 });
200 return transient ? 'pending' : 'skipped';
201}
202
203export async function drainPendingPolarPushes(): Promise<{
204 scanned: number;
205 pushed: number;
206 skipped: number;
207}> {
208 const db = getDb();
209 const rows = await db
210 .select()
211 .from(usageEvents)
212 .where(eq(usageEvents.polarPushStatus, 'pending'))
213 .orderBy(asc(usageEvents.periodStart))
214 .limit(BATCH_SIZE);
215
216 let pushed = 0;
217 let skipped = 0;
218 for (const row of rows) {
219 try {
220 const next = await pushOne(row);
221 if (next === 'pushed') {
222 await db
223 .update(usageEvents)
224 .set({ polarPushStatus: 'pushed', polarPushedAt: new Date() })
225 .where(and(eq(usageEvents.id, row.id)));
226 pushed += 1;
227 } else if (next === 'skipped') {
228 await db
229 .update(usageEvents)
230 .set({ polarPushStatus: 'skipped' })
231 .where(and(eq(usageEvents.id, row.id)));
232 skipped += 1;
233 }
234 // 'pending' = retry next tick — don't touch the row.
235 } catch (err) {
236 log.warn('polar_push_row_failed', {
237 eventId: row.id,
238 message: err instanceof Error ? err.message : String(err),
239 });
240 }
241 }
242
243 if (rows.length > 0) {
244 log.info('polar_push_drain', {
245 scanned: rows.length,
246 pushed,
247 skipped,
248 pending: rows.length - pushed - skipped,
249 });
250 }
251 return { scanned: rows.length, pushed, skipped };
252}
253
254let timer: ReturnType<typeof setInterval> | null = null;
255
256export function startPolarMeterPush(): void {
257 if (timer) return;
258 if (!env.BRIVEN_DATABASE_URL) {
259 log.warn('polar_meter_push_skipped_no_db');
260 return;
261 }
262 // Initial run 30s after boot so migrations finish + the first
263 // aggregator cycle isn't competing for the same rows.
264 setTimeout(() => {
265 void drainPendingPolarPushes();
266 timer = setInterval(() => {
267 void drainPendingPolarPushes();
268 }, INTERVAL_MS);
269 }, 30_000).unref?.();
270 log.info('polar_meter_push_armed', { intervalMs: INTERVAL_MS });
271}