diff --git a/apps/sim/lib/admin/dashboard-organizations.test.ts b/apps/sim/lib/admin/dashboard-organizations.test.ts index 2b6e74f0892..e0780f801a9 100644 --- a/apps/sim/lib/admin/dashboard-organizations.test.ts +++ b/apps/sim/lib/admin/dashboard-organizations.test.ts @@ -233,13 +233,12 @@ describe('listDashboardOrganizations', () => { }, ]) queueTableRows(usageLog, [{ entityId: 'org-1', cost: '2.5' }]) - queueTableRows(usageLog, [{ entityId: 'org-1', workflowRuns: 3 }]) const result = await listDashboardOrganizations({ search: '', limit: 50, offset: 0 }) expect(result.data[0]).toMatchObject({ reportingPeriod: { source: 'stripe' }, - usage: { usedDollars: 2.5, workflowRuns: 3 }, + usage: { usedDollars: 2.5, workflowRuns: null }, }) }) @@ -275,14 +274,13 @@ describe('listDashboardOrganizations', () => { }, ]) queueTableRows(usageLog, [{ entityId: 'org-1', cost: '2.5' }]) - queueTableRows(usageLog, [{ entityId: 'org-1', workflowRuns: 3 }]) queueTableRows(member, [{ organizationId: 'org-1', cost: '1.5' }]) const result = await listDashboardOrganizations({ search: '', limit: 50, offset: 0 }) expect(result.data[0]).toMatchObject({ reportingPeriod: { source: 'reporting', anchorDate: '2026-01-01', interval: 'year' }, - usage: { usedDollars: 2.5, workflowRuns: 3 }, + usage: { usedDollars: 2.5 }, }) }) }) diff --git a/apps/sim/lib/admin/dashboard-usage.integration.ts b/apps/sim/lib/admin/dashboard-usage.integration.ts index 08eab5903af..37caed2d787 100644 --- a/apps/sim/lib/admin/dashboard-usage.integration.ts +++ b/apps/sim/lib/admin/dashboard-usage.integration.ts @@ -4,15 +4,16 @@ import { drizzle } from 'drizzle-orm/postgres-js' import postgres from 'postgres' import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' -const { select } = vi.hoisted(() => ({ select: vi.fn() })) +const { select, transaction } = vi.hoisted(() => ({ select: vi.fn(), transaction: vi.fn() })) const databaseUrl = readTestDatabaseUrl() -vi.mock('@sim/db', () => ({ db: { select } })) +vi.mock('@sim/db', () => ({ db: { select, transaction } })) import { - type DashboardLedgerWindow, - readDashboardHistoricalActorUsage, - readDashboardLedgerUsage, - readDashboardLedgerUsageByUser, + organizationWindow, + readDashboardHistoricalActorSpend, + readDashboardLedgerSpend, + readDashboardLedgerSpendByUser, + readDashboardOrganizationWorkflowRuns, } from '@/lib/admin/dashboard-usage' import type { ResolvedUsagePeriod } from '@/lib/billing/core/reporting-period' @@ -30,11 +31,6 @@ const reportingPeriod: ResolvedUsagePeriod = { anchorDate: '2026-01-15', interval: 'month', } -const organizationWindow = (id: string, period: ResolvedUsagePeriod): DashboardLedgerWindow => ({ - entity: { type: 'organization', id }, - period, -}) - const P = `'2026-08-01', '2026-09-01'` const PREVIOUS = `'2026-07-01', '2026-08-01'` @@ -90,6 +86,9 @@ beforeAll(async () => { ('u3', 'model', 'copilot', 0.1, 'organization', 'org-a', ${P}, NULL, '2026-08-07'), ('u4', 'model', 'copilot', 0.2, 'organization', 'org-a', ${P}, NULL, '2026-08-07'), ('u5', 'model', 'workflow', 0.3, 'organization', 'org-a', ${P}, 'e11', '2026-08-07'), + -- one run shared by two former actors + ('u3', 'model', 'workflow', 0.1, 'organization', 'org-a', ${P}, 'e12', '2026-08-08'), + ('u5', 'model', 'workflow', 0.1, 'organization', 'org-a', ${P}, 'e12', '2026-08-08'), -- org-b, reporting period: matched on created_at, start inclusive and end exclusive, -- whatever the rows are stamped with. @@ -106,6 +105,7 @@ beforeAll(async () => { `) const database = drizzle(connection) select.mockImplementation((fields) => database.select(fields)) + transaction.mockImplementation((run, config) => database.transaction(run, config)) }) afterAll(async () => { @@ -114,63 +114,98 @@ afterAll(async () => { }) describe('dashboard ledger usage', () => { - it('sums spend and counts distinct workflow runs per entity over each period', async () => { - const usage = await readDashboardLedgerUsage([ + it('sums spend per entity over each period', async () => { + const spend = await readDashboardLedgerSpend([ organizationWindow('org-a', stampedPeriod), organizationWindow('org-b', reportingPeriod), organizationWindow('org-c', stampedPeriod), { entity: { type: 'user', id: 'p1' }, period: stampedPeriod }, ]) - expect(Object.fromEntries(usage)).toEqual({ - 'org-a': { cost: 3.38, workflowRuns: 5 }, - 'org-b': { cost: 1.75, workflowRuns: 3 }, - p1: { cost: 0.6, workflowRuns: 1 }, - }) + expect(Object.fromEntries(spend)).toEqual({ 'org-a': 3.58, 'org-b': 1.75, p1: 0.6 }) }) it('reads an entity once however many times it is asked for', async () => { - const usage = await readDashboardLedgerUsage([ + const spend = await readDashboardLedgerSpend([ organizationWindow('org-a', stampedPeriod), organizationWindow('org-a', stampedPeriod), ]) - expect(usage.get('org-a')).toEqual({ cost: 3.38, workflowRuns: 5 }) + expect(spend.get('org-a')).toBe(3.58) }) - it('splits spend and runs by acting user, counting a shared run for each actor', async () => { - const usage = await readDashboardLedgerUsageByUser([ + it('splits spend by acting user', async () => { + const spend = await readDashboardLedgerSpendByUser([ organizationWindow('org-a', stampedPeriod), organizationWindow('org-b', reportingPeriod), ]) - expect(Object.fromEntries(usage.get('org-a') ?? [])).toEqual({ - u1: { cost: 1.51, workflowRuns: 2 }, - u2: { cost: 0.37, workflowRuns: 2 }, - u3: { cost: 1, workflowRuns: 1 }, - u4: { cost: 0.2, workflowRuns: 0 }, - u5: { cost: 0.3, workflowRuns: 1 }, - }) - expect(Object.fromEntries(usage.get('org-b') ?? [])).toEqual({ - u6: { cost: 1.75, workflowRuns: 3 }, + expect(Object.fromEntries(spend.get('org-a') ?? [])).toEqual({ + u1: 1.51, + u2: 0.37, + u3: 1.1, + u4: 0.2, + u5: 0.4, }) + expect(Object.fromEntries(spend.get('org-b') ?? [])).toEqual({ u6: 1.75 }) }) it('narrows the per-user split to the requested users', async () => { const window = organizationWindow('org-a', stampedPeriod) - const usage = await readDashboardLedgerUsageByUser([window], ['u2', 'u4']) - expect(Object.fromEntries(usage.get('org-a') ?? [])).toEqual({ - u2: { cost: 0.37, workflowRuns: 2 }, - u4: { cost: 0.2, workflowRuns: 0 }, - }) - expect((await readDashboardLedgerUsageByUser([window], [])).size).toBe(0) + const spend = await readDashboardLedgerSpendByUser([window], ['u2', 'u4']) + expect(Object.fromEntries(spend.get('org-a') ?? [])).toEqual({ u2: 0.37, u4: 0.2 }) + expect((await readDashboardLedgerSpendByUser([window], [])).size).toBe(0) }) - it('counts only actors who are neither members nor collaborators of the organization', async () => { - expect(await readDashboardHistoricalActorUsage('org-a', stampedPeriod)).toEqual({ - usedDollars: 1.5, - workflowRuns: 2, - actorCount: 3, + it('counts distinct workflow runs in total, for each actor of a shared run, and once across former actors', async () => { + const orgA = await readDashboardOrganizationWorkflowRuns('org-a', stampedPeriod) + expect(orgA?.total).toBe(6) + expect(Object.fromEntries(orgA?.byUser ?? [])).toEqual({ u1: 2, u2: 2, u3: 2, u5: 2 }) + expect(orgA?.formerActors).toBe(3) + + const orgB = await readDashboardOrganizationWorkflowRuns('org-b', reportingPeriod) + expect(orgB?.total).toBe(3) + expect(Object.fromEntries(orgB?.byUser ?? [])).toEqual({ u6: 3 }) + expect(orgB?.formerActors).toBe(3) + + const empty = await readDashboardOrganizationWorkflowRuns('org-c', stampedPeriod) + expect(empty?.total).toBe(0) + expect(empty?.byUser.size).toBe(0) + expect(empty?.formerActors).toBe(0) + }) + + it('reports runs as unknown when the count outlasts its budget', async () => { + const locker = postgres(databaseUrl, { + max: 1, + prepare: false, + connection: { search_path: schemaName }, + onnotice: () => undefined, + }) + let release: () => void = () => undefined + const released = new Promise((resolve) => { + release = resolve + }) + let markLocked: () => void = () => undefined + const locked = new Promise((resolve) => { + markLocked = resolve + }) + const holder = locker.begin(async (tx) => { + await tx.unsafe('LOCK TABLE usage_log IN ACCESS EXCLUSIVE MODE') + markLocked() + await released }) + try { + await locked + expect(await readDashboardOrganizationWorkflowRuns('org-a', stampedPeriod)).toBeNull() + } finally { + release() + await holder + await locker.end() + } + }, 20_000) + + it('spends only for actors who are neither members nor collaborators of the organization', async () => { + const spend = await readDashboardHistoricalActorSpend('org-a', stampedPeriod) + expect(Object.fromEntries(spend)).toEqual({ u3: 1.1, u4: 0.2, u5: 0.4 }) }) }) diff --git a/apps/sim/lib/admin/dashboard-usage.ts b/apps/sim/lib/admin/dashboard-usage.ts index ac37df8722a..f76a77a062d 100644 --- a/apps/sim/lib/admin/dashboard-usage.ts +++ b/apps/sim/lib/admin/dashboard-usage.ts @@ -1,24 +1,33 @@ import { db } from '@sim/db' import { member, permissions, usageLog, workspace } from '@sim/db/schema' -import { and, countDistinct, eq, inArray, notExists, type SQL, sql } from 'drizzle-orm' -import { alias, unionAll } from 'drizzle-orm/pg-core' +import { createLogger } from '@sim/logger' +import { getPostgresCancellationReason } from '@sim/utils/errors' +import { and, countDistinct, eq, inArray, notInArray, type SQL, sql } from 'drizzle-orm' +import { unionAll } from 'drizzle-orm/pg-core' +import { readLedgerBounded } from '@/lib/billing/core/ledger-read' import type { ResolvedUsagePeriod } from '@/lib/billing/core/reporting-period' import { buildUsageAnalyticsScope } from '@/lib/billing/core/usage-analytics' import type { BillingEntity } from '@/lib/billing/core/usage-log' +import { DatabaseReadDeadlineError } from '@/lib/db/read-retry' /** * Ledger reads behind the admin dashboard: each billing entity's spend and workflow runs * over its current usage period. * - * Each read is one statement whose `UNION ALL` arms each cover one entity, so every arm - * is a scan bounded by one entity's period on `usage_log_billing_entity_period_cost_idx` - * (stamped periods) or `usage_log_billing_entity_created_at_cost_idx` (reporting periods). - * One `BitmapOr` over all of them would fetch every matched row from the heap instead. + * Spend is index-only, so the list pages read it for a whole page of entities: one + * statement whose `UNION ALL` arms each cover one entity, so every arm is a scan bounded by + * one entity's period on `usage_log_billing_entity_period_cost_idx` (stamped periods) or + * `usage_log_billing_entity_created_at_cost_idx` (reporting periods). One `BitmapOr` over + * all of them would fetch every matched row from the heap instead. * - * Spend is index-only. Workflow runs are a separate statement because no index carries - * `execution_id`: they fetch each matched workflow row from the heap, and keeping them - * apart leaves the spend scans index-only. An index for them would cost every workflow ledger - * insert an extra write to speed up a staff-only page. + * Workflow runs are not. Neither entity index carries `execution_id`, so counting them + * fetches every matched workflow row from the heap, and a large payer's period is millions + * of rows scattered across it: a page of entities, or even one, can outlast the request. + * Widening both entity indexes with it would grow the hot billing sums' working set to + * speed up a staff-only page. So only one organization's detail counts runs, in a single + * scan that yields the total and every actor's share, under + * {@link WORKFLOW_RUNS_BUDGET_MS}; past it the count is unknown (`null`) and the page + * still loads. * * They stay on the primary, where the hot billing sums keep those index pages cached. On * the replica the same scans read from disk, and a statement held past the standby's @@ -30,18 +39,36 @@ import type { BillingEntity } from '@/lib/billing/core/usage-log' * overcount runs, and only the ledger says which entity a run was billed to. */ -export interface DashboardLedgerWindow { +const logger = createLogger('AdminDashboardUsage') + +interface DashboardLedgerWindow { entity: BillingEntity period: ResolvedUsagePeriod } -export interface DashboardLedgerUsage { - cost: number - workflowRuns: number +interface DashboardWorkflowRuns { + total: number + /** Runs per acting user; a run with rows from two actors counts for each. */ + byUser: Map + /** Distinct runs across the organization's former actors together. */ + formerActors: number } +/** + * How long one detail page may spend counting workflow runs. Well inside the admin + * proxy's request timeout, so a payer too large to count still gets its page. + */ +const WORKFLOW_RUNS_BUDGET_MS = 5_000 + const WORKFLOW_SOURCE = sql`${usageLog.source} = 'workflow'` +export function organizationWindow( + organizationId: string, + period: ResolvedUsagePeriod +): DashboardLedgerWindow { + return { entity: { type: 'organization', id: organizationId }, period } +} + function windowScope({ entity, period }: DashboardLedgerWindow): SQL[] { return buildUsageAnalyticsScope(entity, { kind: 'period', period }) } @@ -56,13 +83,13 @@ function distinctWindows(windows: DashboardLedgerWindow[]): DashboardLedgerWindo } /** - * The spend rows and the workflow-run rows over every scope, keyed by entity and, when - * `byUser`, by acting user. Each is one statement with an arm per scope. + * The spend rows over every scope, keyed by entity and, when `byUser`, by acting user. + * One statement with an arm per scope. */ -async function readLedgerRows(scopes: SQL[][], byUser: boolean) { +function readSpendRows(scopes: SQL[][], byUser: boolean) { const userId = byUser ? sql`${usageLog.userId}` : sql`''` const groupBy = byUser ? [usageLog.billingEntityId, usageLog.userId] : [usageLog.billingEntityId] - const [cost, secondCost, ...moreCost] = scopes.map((scope) => + const [first, second, ...more] = scopes.map((scope) => db .select({ entityId: usageLog.billingEntityId, @@ -73,143 +100,136 @@ async function readLedgerRows(scopes: SQL[][], byUser: boolean) { .where(and(...scope)) .groupBy(...groupBy) ) - const [runs, secondRuns, ...moreRuns] = scopes.map((scope) => - db - .select({ - entityId: usageLog.billingEntityId, - userId, - workflowRuns: countDistinct(usageLog.executionId), - }) - .from(usageLog) - .where(and(...scope, WORKFLOW_SOURCE)) - .groupBy(...groupBy) - ) - return Promise.all([ - secondCost ? unionAll(cost, secondCost, ...moreCost) : cost, - secondRuns ? unionAll(runs, secondRuns, ...moreRuns) : runs, - ]) -} - -function usageFor(map: Map, key: K): DashboardLedgerUsage { - let usage = map.get(key) - if (!usage) { - usage = { cost: 0, workflowRuns: 0 } - map.set(key, usage) - } - return usage + return second ? unionAll(first, second, ...more) : first } -/** Spend and workflow runs per entity id. An entity with no ledger rows is absent. */ -export async function readDashboardLedgerUsage( +/** Spend per entity id. An entity with no ledger rows is absent. */ +export async function readDashboardLedgerSpend( windows: DashboardLedgerWindow[] -): Promise> { - const result = new Map() +): Promise> { + const result = new Map() const scopes = distinctWindows(windows).map(windowScope) if (scopes.length === 0) return result - const [costRows, runRows] = await readLedgerRows(scopes, false) - for (const row of costRows) { - if (row.entityId) usageFor(result, row.entityId).cost += Number(row.cost) - } - for (const row of runRows) { - if (row.entityId) usageFor(result, row.entityId).workflowRuns += row.workflowRuns + for (const row of await readSpendRows(scopes, false)) { + if (row.entityId) result.set(row.entityId, Number(row.cost)) } return result } -/** - * Spend and workflow runs per entity id, then per acting user, narrowed to `userIds` - * when given. A run is counted for each user with a row in it. - */ -export async function readDashboardLedgerUsageByUser( +/** Spend per entity id, then per acting user, narrowed to `userIds` when given. */ +export async function readDashboardLedgerSpendByUser( windows: DashboardLedgerWindow[], userIds?: string[] -): Promise>> { - const result = new Map>() +): Promise>> { + const result = new Map>() const scopes = distinctWindows(windows).map((window) => userIds ? [...windowScope(window), inArray(usageLog.userId, userIds)] : windowScope(window) ) if (scopes.length === 0 || userIds?.length === 0) return result - const usersOf = (entityId: string) => { - let users = result.get(entityId) + for (const row of await readSpendRows(scopes, true)) { + if (!row.entityId) continue + let users = result.get(row.entityId) if (!users) { users = new Map() - result.set(entityId, users) - } - return users - } - const [costRows, runRows] = await readLedgerRows(scopes, true) - for (const row of costRows) { - if (row.entityId) usageFor(usersOf(row.entityId), row.userId).cost += Number(row.cost) - } - for (const row of runRows) { - if (row.entityId) { - usageFor(usersOf(row.entityId), row.userId).workflowRuns += row.workflowRuns + result.set(row.entityId, users) } + users.set(row.userId, Number(row.cost)) } return result } -const historicalUsageMember = alias(member, 'historical_usage_member') -const historicalUsagePermission = alias(permissions, 'historical_usage_permission') -const historicalUsageWorkspace = alias(workspace, 'historical_usage_workspace') +/** + * Who no longer represents an organization's usage: an actor who is neither a member nor a + * collaborator on one of its workspaces any more. Uncorrelated, so each list is hashed once + * rather than probed for every ledger row. + */ +function isFormerActor(organizationId: string): SQL { + const members = db + .select({ userId: member.userId }) + .from(member) + .where(eq(member.organizationId, organizationId)) + const collaborators = db + .select({ userId: permissions.userId }) + .from(permissions) + .innerJoin( + workspace, + and(eq(permissions.entityType, 'workspace'), eq(permissions.entityId, workspace.id)) + ) + .where(eq(workspace.organizationId, organizationId)) + return sql`(${notInArray(usageLog.userId, members)} and ${notInArray(usageLog.userId, collaborators)})` +} /** - * An organization's period usage by actors who are neither members nor collaborators on - * one of its workspaces any more: spend, workflow runs and how many such actors. + * An organization's workflow runs over its period, in total, per acting user and across its + * former actors, from one scan. `null` when the count does not finish within + * {@link WORKFLOW_RUNS_BUDGET_MS}. */ -export async function readDashboardHistoricalActorUsage( +export async function readDashboardOrganizationWorkflowRuns( organizationId: string, period: ResolvedUsagePeriod -): Promise<{ usedDollars: number; workflowRuns: number; actorCount: number }> { - const currentMember = db - .select({ value: sql`1` }) - .from(historicalUsageMember) - .where( - and( - eq(historicalUsageMember.organizationId, organizationId), - eq(historicalUsageMember.userId, usageLog.userId) - ) +): Promise { + let rows: { userId: string | null; formerActor: boolean | null; workflowRuns: number }[] + try { + rows = await readLedgerBounded( + db, + (tx) => { + const workflowRows = tx + .select({ + userId: usageLog.userId, + executionId: usageLog.executionId, + formerActor: sql`${isFormerActor(organizationId)}`.as('former_actor'), + }) + .from(usageLog) + .where(and(...windowScope(organizationWindow(organizationId, period)), WORKFLOW_SOURCE)) + .as('workflow_rows') + return tx + .select({ + userId: sql`${workflowRows.userId}`, + formerActor: sql`${workflowRows.formerActor}`, + workflowRuns: countDistinct(workflowRows.executionId), + }) + .from(workflowRows) + .groupBy(sql`grouping sets ((${workflowRows.userId}), (${workflowRows.formerActor}), ())`) + }, + WORKFLOW_RUNS_BUDGET_MS ) - const currentCollaborator = db - .select({ value: sql`1` }) - .from(historicalUsagePermission) - .innerJoin( - historicalUsageWorkspace, - and( - eq(historicalUsagePermission.entityType, 'workspace'), - eq(historicalUsagePermission.entityId, historicalUsageWorkspace.id) - ) - ) - .where( - and( - eq(historicalUsageWorkspace.organizationId, organizationId), - eq(historicalUsagePermission.userId, usageLog.userId) - ) - ) - const scope = [ - ...windowScope({ entity: { type: 'organization', id: organizationId }, period }), - notExists(currentMember), - notExists(currentCollaborator), - ] + } catch (error) { + const overBudget = + error instanceof DatabaseReadDeadlineError || + getPostgresCancellationReason(error) === 'statement_timeout' + if (!overBudget) throw error + logger.warn('Workflow run count exceeded its budget', { + organizationId, + budgetMs: WORKFLOW_RUNS_BUDGET_MS, + }) + return null + } - const [[costRow], [runRow]] = await Promise.all([ - db - .select({ - usedDollars: sql`coalesce(sum(${usageLog.cost}), 0)`, - actorCount: countDistinct(usageLog.userId), - }) - .from(usageLog) - .where(and(...scope)), - db - .select({ workflowRuns: countDistinct(usageLog.executionId) }) - .from(usageLog) - .where(and(...scope, WORKFLOW_SOURCE)), - ]) - return { - usedDollars: Number(costRow?.usedDollars ?? 0), - workflowRuns: runRow?.workflowRuns ?? 0, - actorCount: costRow?.actorCount ?? 0, + const runs: DashboardWorkflowRuns = { total: 0, byUser: new Map(), formerActors: 0 } + for (const row of rows) { + if (row.userId !== null) runs.byUser.set(row.userId, row.workflowRuns) + else if (row.formerActor === null) runs.total = row.workflowRuns + else if (row.formerActor) runs.formerActors = row.workflowRuns } + return runs +} + +/** An organization's period spend per former actor (see {@link isFormerActor}). */ +export async function readDashboardHistoricalActorSpend( + organizationId: string, + period: ResolvedUsagePeriod +): Promise> { + const rows = await db + .select({ + userId: usageLog.userId, + cost: sql`coalesce(sum(${usageLog.cost}), 0)`, + }) + .from(usageLog) + .where( + and(...windowScope(organizationWindow(organizationId, period)), isFormerActor(organizationId)) + ) + .groupBy(usageLog.userId) + return new Map(rows.map((row) => [row.userId, Number(row.cost)])) } diff --git a/apps/sim/lib/admin/dashboard.ts b/apps/sim/lib/admin/dashboard.ts index bdb6add9ba2..252bfd2ad9b 100644 --- a/apps/sim/lib/admin/dashboard.ts +++ b/apps/sim/lib/admin/dashboard.ts @@ -15,9 +15,11 @@ import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { and, count, countDistinct, desc, eq, ilike, inArray, isNull, or, sql } from 'drizzle-orm' import { - readDashboardHistoricalActorUsage, - readDashboardLedgerUsage, - readDashboardLedgerUsageByUser, + organizationWindow, + readDashboardHistoricalActorSpend, + readDashboardLedgerSpend, + readDashboardLedgerSpendByUser, + readDashboardOrganizationWorkflowRuns, } from '@/lib/admin/dashboard-usage' import { getOrganizationUsageLimitFallbackDollars, @@ -166,87 +168,17 @@ interface DashboardOrganizationSummaryInput { provisioning: EnterpriseProvisioningView | null owner: { id: string; name: string; email: string } | null usageDollars: number - workflowRuns: number + /** `null` where runs are not counted: list pages, and a detail past its run budget. */ + workflowRuns: number | null usagePeriod: ResolvedUsagePeriod } -interface DashboardOrganizationUsageContext { - organizationId: string - period: ResolvedUsagePeriod -} - -interface DashboardOrganizationUsage { - total: number - byUser: Map - workflowRuns: number - workflowRunsByUser: Map -} - function resolveDashboardUsagePeriod( latestSubscription: typeof subscription.$inferSelect | null ): ResolvedUsagePeriod { return resolveSubscriptionUsagePeriodOrDefault(latestSubscription ?? {}) } -async function getDashboardOrganizationUsage( - contexts: DashboardOrganizationUsageContext[], - options: { includeUserBreakdown?: boolean; userIds?: string[] } = {} -): Promise> { - const result = new Map() - for (const context of contexts) { - result.set(context.organizationId, { - total: 0, - byUser: new Map(), - workflowRuns: 0, - workflowRunsByUser: new Map(), - }) - } - if (contexts.length === 0) return result - - const includeUserBreakdown = options.includeUserBreakdown ?? true - if (options.userIds?.length === 0) return result - const windows = contexts.map(({ organizationId, period }) => ({ - entity: { type: 'organization' as const, id: organizationId }, - period, - })) - - if (!includeUserBreakdown) { - for (const [organizationId, ledger] of await readDashboardLedgerUsage(windows)) { - const usage = result.get(organizationId) - if (usage) { - usage.total += ledger.cost - usage.workflowRuns += ledger.workflowRuns - } - } - return result - } - - const ledgerByUser = await readDashboardLedgerUsageByUser(windows, options.userIds) - for (const [organizationId, users] of ledgerByUser) { - const usage = result.get(organizationId) - if (!usage) continue - for (const [userId, ledger] of users) { - usage.total += ledger.cost - usage.byUser.set(userId, (usage.byUser.get(userId) ?? 0) + ledger.cost) - usage.workflowRuns += ledger.workflowRuns - usage.workflowRunsByUser.set( - userId, - (usage.workflowRunsByUser.get(userId) ?? 0) + ledger.workflowRuns - ) - } - } - - return result -} - -async function getHistoricalActorUsage( - organizationId: string, - period: ResolvedUsagePeriod -): Promise<{ usedDollars: number; usedCredits: number; workflowRuns: number; actorCount: number }> { - const usage = await readDashboardHistoricalActorUsage(organizationId, period) - return { ...usage, usedCredits: dollarsToCredits(usage.usedDollars) } -} - export function toDashboardProvisioning(view: EnterpriseProvisioningView) { const { usageLimitCredits, ...rest } = view return { @@ -450,31 +382,32 @@ export async function listDashboardUsers({ search, limit, offset }: PaginationIn const organizationSubscriptionMap = new Map( organizationSubscriptions.map((row) => [row.referenceId, row]) ) - const organizationUsage = await getDashboardOrganizationUsage( - organizationIds.map((organizationId) => ({ - organizationId, - period: resolveDashboardUsagePeriod(organizationSubscriptionMap.get(organizationId) ?? null), - })), - { userIds: rows.map((row) => row.id) } - ) const personalSubscriptionMap = new Map( personalSubscriptions.map((row) => [row.referenceId, row]) ) - const personalUsage = await readDashboardLedgerUsage( - personalUserIds.map((userId) => ({ - entity: { type: 'user' as const, id: userId }, - period: resolveDashboardUsagePeriod(personalSubscriptionMap.get(userId) ?? null), - })) - ) + const [organizationSpend, personalSpend] = await Promise.all([ + readDashboardLedgerSpendByUser( + organizationIds.map((organizationId) => + organizationWindow( + organizationId, + resolveDashboardUsagePeriod(organizationSubscriptionMap.get(organizationId) ?? null) + ) + ), + rows.map((row) => row.id) + ), + readDashboardLedgerSpend( + personalUserIds.map((userId) => ({ + entity: { type: 'user' as const, id: userId }, + period: resolveDashboardUsagePeriod(personalSubscriptionMap.get(userId) ?? null), + })) + ), + ]) return { data: rows.map((row) => { - const organization = row.organizationId - ? organizationUsage.get(row.organizationId) - : undefined const usageDollars = row.organizationId - ? (organization?.byUser.get(row.id) ?? 0) - : (personalUsage.get(row.id)?.cost ?? 0) + ? (organizationSpend.get(row.organizationId)?.get(row.id) ?? 0) + : (personalSpend.get(row.id) ?? 0) return { id: row.id, name: row.name, @@ -485,9 +418,7 @@ export async function listDashboardUsers({ search, limit, offset }: PaginationIn : null, usageDollars, usageCredits: dollarsToCredits(usageDollars), - workflowRuns: row.organizationId - ? (organization?.workflowRunsByUser.get(row.id) ?? 0) - : (personalUsage.get(row.id)?.workflowRuns ?? 0), + workflowRuns: null, } }), pagination: { @@ -527,27 +458,26 @@ async function getDashboardOrganizationSummary(organizationId: string) { ]) if (!org) return null - const [owner] = await db - .select({ id: user.id, name: user.name, email: user.email }) - .from(member) - .innerJoin(user, eq(user.id, member.userId)) - .where(and(eq(member.organizationId, organizationId), eq(member.role, 'owner'))) - .limit(1) - const memberCount = memberCountRow?.value ?? 0 const period = resolveDashboardUsagePeriod(latestSubscription) - const usage = await getDashboardOrganizationUsage([{ organizationId, period }], { - includeUserBreakdown: false, - }) + const [[owner], spend] = await Promise.all([ + db + .select({ id: user.id, name: user.name, email: user.email }) + .from(member) + .innerJoin(user, eq(user.id, member.userId)) + .where(and(eq(member.organizationId, organizationId), eq(member.role, 'owner'))) + .limit(1), + readDashboardLedgerSpend([organizationWindow(organizationId, period)]), + ]) return { ...buildDashboardOrganizationSummary({ org, - memberCount, + memberCount: memberCountRow?.value ?? 0, externalCollaboratorCount: externalCountRow?.value ?? 0, latestSubscription, provisioning: provisionings.get(organizationId) ?? null, owner: owner ?? null, - usageDollars: usage.get(organizationId)?.total ?? 0, - workflowRuns: usage.get(organizationId)?.workflowRuns ?? 0, + usageDollars: spend.get(organizationId) ?? 0, + workflowRuns: null, usagePeriod: period, }), usagePeriod: period, @@ -634,23 +564,14 @@ export async function listDashboardOrganizations({ search, limit, offset }: Pagi ) ) const subscriptionByOrganization = new Map(subscriptionRows.map((row) => [row.referenceId, row])) - const usagePeriodsByOrganization = new Map( - organizationIds.map( - (organizationId) => - [ - organizationId, - resolveDashboardUsagePeriod(subscriptionByOrganization.get(organizationId) ?? null), - ] as const - ) - ) - const usageByOrganization = await getDashboardOrganizationUsage( - organizationIds.map((organizationId) => ({ - organizationId, - period: usagePeriodsByOrganization.get(organizationId)!, - })), - { includeUserBreakdown: false } + const organizations = orgRows.map((org) => { + const latestSubscription = subscriptionByOrganization.get(org.id) ?? null + return { org, latestSubscription, usagePeriod: resolveDashboardUsagePeriod(latestSubscription) } + }) + const spendByOrganization = await readDashboardLedgerSpend( + organizations.map(({ org, usagePeriod }) => organizationWindow(org.id, usagePeriod)) ) - const data = orgRows.map((org) => { + const data = organizations.map(({ org, latestSubscription, usagePeriod }) => { const membership = membershipsByOrganization.get(org.id) const owner = membership?.ownerId && membership.ownerName && membership.ownerEmail @@ -664,12 +585,12 @@ export async function listDashboardOrganizations({ search, limit, offset }: Pagi org, memberCount: membership?.memberCount ?? 0, externalCollaboratorCount: externalCountByOrganization.get(org.id) ?? 0, - latestSubscription: subscriptionByOrganization.get(org.id) ?? null, + latestSubscription, provisioning: provisionings.get(org.id) ?? null, owner, - usageDollars: usageByOrganization.get(org.id)?.total ?? 0, - workflowRuns: usageByOrganization.get(org.id)?.workflowRuns ?? 0, - usagePeriod: usagePeriodsByOrganization.get(org.id)!, + usageDollars: spendByOrganization.get(org.id) ?? 0, + workflowRuns: null, + usagePeriod, }) return summary }) @@ -746,91 +667,116 @@ export async function getDashboardOrganization( .where(eq(workspace.organizationId, organizationId)) .orderBy(workspace.name, workspace.id) - const [memberRows, externalRows, workspaceRows, workspaceCountRows, configurationIntent] = - await Promise.all([ - memberQuery().limit(pagination.limit).offset(pagination.memberOffset), - externalCollaboratorQuery() - .limit(pagination.limit) - .offset(pagination.externalCollaboratorOffset), - workspaceQuery().limit(pagination.limit).offset(pagination.workspaceOffset), - db - .select({ value: count() }) - .from(workspace) - .where(eq(workspace.organizationId, organizationId)), - subscriptionRow?.plan === 'enterprise' - ? resolveEnterpriseMetadataIntent(db, subscriptionRow.id, subscriptionRow.metadata) - : Promise.resolve(null), + const window = organizationWindow(organizationId, usagePeriod) + const readPage = async () => { + const [memberRows, externalRows, workspaceRows, workspaceCountRows, configurationIntent] = + await Promise.all([ + memberQuery().limit(pagination.limit).offset(pagination.memberOffset), + externalCollaboratorQuery() + .limit(pagination.limit) + .offset(pagination.externalCollaboratorOffset), + workspaceQuery().limit(pagination.limit).offset(pagination.workspaceOffset), + db + .select({ value: count() }) + .from(workspace) + .where(eq(workspace.organizationId, organizationId)), + subscriptionRow?.plan === 'enterprise' + ? resolveEnterpriseMetadataIntent(db, subscriptionRow.id, subscriptionRow.metadata) + : Promise.resolve(null), + ]) + const visibleUserIds = [ + ...new Set([ + ...memberRows.map((row) => row.userId), + ...externalRows.map((row) => row.userId), + ]), + ] + const [limitRows, spendByOrganization, historicalActorSpend] = await Promise.all([ + visibleUserIds.length === 0 + ? Promise.resolve([]) + : db + .select({ + userId: organizationMemberUsageLimit.userId, + limit: organizationMemberUsageLimit.usageLimit, + }) + .from(organizationMemberUsageLimit) + .where( + and( + eq(organizationMemberUsageLimit.organizationId, organizationId), + inArray(organizationMemberUsageLimit.userId, visibleUserIds) + ) + ), + readDashboardLedgerSpendByUser([window], visibleUserIds), + readDashboardHistoricalActorSpend(organizationId, usagePeriod), ]) - const visibleUserIds = [ - ...new Set([...memberRows.map((row) => row.userId), ...externalRows.map((row) => row.userId)]), - ] - const [limitRows, usageByOrganization, historicalActorUsage] = await Promise.all([ - visibleUserIds.length === 0 - ? Promise.resolve([]) - : db - .select({ - userId: organizationMemberUsageLimit.userId, - limit: organizationMemberUsageLimit.usageLimit, - }) - .from(organizationMemberUsageLimit) - .where( - and( - eq(organizationMemberUsageLimit.organizationId, organizationId), - inArray(organizationMemberUsageLimit.userId, visibleUserIds) - ) - ), - getDashboardOrganizationUsage([{ organizationId, period: usagePeriod }], { - userIds: visibleUserIds, - }), - getHistoricalActorUsage(organizationId, usagePeriod), + return { + memberRows, + externalRows, + workspaceRows, + workspaceCountRows, + configurationIntent, + limitRows, + spendByOrganization, + historicalActorSpend, + } + } + const [workflowRuns, page] = await Promise.all([ + readDashboardOrganizationWorkflowRuns(organizationId, usagePeriod), + readPage(), ]) - const limits = new Map(limitRows.map((row) => [row.userId, Number(row.limit)])) - const organizationUsage = usageByOrganization.get(organizationId) - const usageByUser = organizationUsage?.byUser ?? new Map() - const workflowRunsByUser = organizationUsage?.workflowRunsByUser ?? new Map() - const workspaceTotal = workspaceCountRows[0]?.value ?? 0 + const limits = new Map(page.limitRows.map((row) => [row.userId, Number(row.limit)])) + const usageByUser = page.spendByOrganization.get(organizationId) ?? new Map() + const runsOf = (userId: string) => (workflowRuns ? (workflowRuns.byUser.get(userId) ?? 0) : null) + let historicalUsedDollars = 0 + for (const cost of page.historicalActorSpend.values()) historicalUsedDollars += cost + const workspaceTotal = page.workspaceCountRows[0]?.value ?? 0 return { ...base, + usage: { ...base.usage, workflowRuns: workflowRuns?.total ?? null }, configurationUpdate: toDashboardConfigurationUpdate( - configurationIntent, + page.configurationIntent, base.prepaidBalanceDollars ), - historicalActorUsage, - members: memberRows.map((row) => ({ + historicalActorUsage: { + usedDollars: historicalUsedDollars, + usedCredits: dollarsToCredits(historicalUsedDollars), + workflowRuns: workflowRuns?.formerActors ?? null, + actorCount: page.historicalActorSpend.size, + }, + members: page.memberRows.map((row) => ({ ...row, usageLimitDollars: limits.get(row.userId) ?? null, usageDollars: usageByUser.get(row.userId) ?? 0, usageCredits: dollarsToCredits(usageByUser.get(row.userId) ?? 0), - workflowRuns: workflowRunsByUser.get(row.userId) ?? 0, + workflowRuns: runsOf(row.userId), })), - externalCollaborators: externalRows.map((row) => ({ + externalCollaborators: page.externalRows.map((row) => ({ ...row, workspaceCount: row.workspaceCount, usageLimitDollars: limits.get(row.userId) ?? null, usageDollars: usageByUser.get(row.userId) ?? 0, usageCredits: dollarsToCredits(usageByUser.get(row.userId) ?? 0), - workflowRuns: workflowRunsByUser.get(row.userId) ?? 0, + workflowRuns: runsOf(row.userId), })), - workspaces: workspaceRows, + workspaces: page.workspaceRows, memberPagination: { total: base.memberCount, limit: pagination.limit, offset: pagination.memberOffset, - hasMore: pagination.memberOffset + memberRows.length < base.memberCount, + hasMore: pagination.memberOffset + page.memberRows.length < base.memberCount, }, externalCollaboratorPagination: { total: base.externalCollaboratorCount, limit: pagination.limit, offset: pagination.externalCollaboratorOffset, hasMore: - pagination.externalCollaboratorOffset + externalRows.length < + pagination.externalCollaboratorOffset + page.externalRows.length < base.externalCollaboratorCount, }, workspacePagination: { total: workspaceTotal, limit: pagination.limit, offset: pagination.workspaceOffset, - hasMore: pagination.workspaceOffset + workspaceRows.length < workspaceTotal, + hasMore: pagination.workspaceOffset + page.workspaceRows.length < workspaceTotal, }, subscription: subscriptionRow ? { @@ -978,11 +924,10 @@ export async function previewDashboardEnterpriseReportingPeriod( .limit(1), ]) if (!org || !subscriptionRow) throw new Error('Active Enterprise subscription not found') - const usage = await getDashboardOrganizationUsage([{ organizationId, period: reportingPeriod }], { - includeUserBreakdown: false, - }) - const usedDollars = usage.get(organizationId)?.total ?? 0 - const workflowRuns = usage.get(organizationId)?.workflowRuns ?? 0 + const spend = await readDashboardLedgerSpend([ + organizationWindow(organizationId, reportingPeriod), + ]) + const usedDollars = spend.get(organizationId) ?? 0 const limitDollars = Number(org.orgUsageLimit ?? 0) return { reportingPeriod: { @@ -997,7 +942,7 @@ export async function previewDashboardEnterpriseReportingPeriod( limitDollars, usedCredits: dollarsToCredits(usedDollars), limitCredits: dollarsToCredits(limitDollars), - workflowRuns, + workflowRuns: null, }, exceedsLimit: usedDollars > limitDollars, } diff --git a/apps/sim/lib/api/contracts/v1/admin/dashboard.ts b/apps/sim/lib/api/contracts/v1/admin/dashboard.ts index 4df4029cc7f..26ae84dde19 100644 --- a/apps/sim/lib/api/contracts/v1/admin/dashboard.ts +++ b/apps/sim/lib/api/contracts/v1/admin/dashboard.ts @@ -28,6 +28,12 @@ const positiveCreditAlignedDollarAmountSchema = creditAlignedDollarAmountSchema. } ) +/** + * Billable workflow runs over a usage period, or `null` where they are not counted: + * list pages, the reporting-period preview, and an organization detail past its run budget. + */ +const adminDashboardWorkflowRunsSchema = z.number().int().min(0).nullable() + export const adminDashboardUserSchema = z.object({ id: z.string(), name: z.string(), @@ -35,7 +41,7 @@ export const adminDashboardUserSchema = z.object({ activeOrganization: z.object({ id: z.string(), name: z.string() }).nullable(), usageDollars: dollarAmountSchema, usageCredits: z.number().int().min(0), - workflowRuns: z.number().int().min(0), + workflowRuns: adminDashboardWorkflowRunsSchema, }) const adminDashboardBillingIntervalSchema = z.enum(['month', 'year']) @@ -51,7 +57,7 @@ const adminDashboardUsageSchema = z.object({ limitDollars: dollarAmountSchema, usedCredits: z.number().int().min(0), limitCredits: z.number().int().min(0), - workflowRuns: z.number().int().min(0), + workflowRuns: adminDashboardWorkflowRunsSchema, }) const adminDashboardWorkspaceMoveProgressSchema = z.object({ selected: z.number().int().min(0), @@ -184,7 +190,7 @@ export const adminDashboardOrganizationDetailSchema = historicalActorUsage: z.object({ usedDollars: dollarAmountSchema, usedCredits: z.number().int().min(0), - workflowRuns: z.number().int().min(0), + workflowRuns: adminDashboardWorkflowRunsSchema, actorCount: z.number().int().min(0), }), members: z.array( @@ -197,7 +203,7 @@ export const adminDashboardOrganizationDetailSchema = usageLimitDollars: dollarAmountSchema.nullable(), usageDollars: dollarAmountSchema, usageCredits: z.number().int().min(0), - workflowRuns: z.number().int().min(0), + workflowRuns: adminDashboardWorkflowRunsSchema, }) ), externalCollaborators: z.array( @@ -209,7 +215,7 @@ export const adminDashboardOrganizationDetailSchema = usageLimitDollars: dollarAmountSchema.nullable(), usageDollars: dollarAmountSchema, usageCredits: z.number().int().min(0), - workflowRuns: z.number().int().min(0), + workflowRuns: adminDashboardWorkflowRunsSchema, }) ), workspaces: z.array(z.object({ id: z.string(), name: z.string() })), diff --git a/apps/sim/lib/billing/core/ledger-read.ts b/apps/sim/lib/billing/core/ledger-read.ts index ef1b70cf097..78ed1cd5a73 100644 --- a/apps/sim/lib/billing/core/ledger-read.ts +++ b/apps/sim/lib/billing/core/ledger-read.ts @@ -16,16 +16,16 @@ import type { DbClient, DbTransaction } from '@/lib/db/types' */ export function readLedgerBounded( executor: DbClient, - read: (tx: DbTransaction) => Promise + read: (tx: DbTransaction) => Promise, + /** A tighter bound for a read that can do without its answer, such as a staff-only count. */ + budgetMs: number = USAGE_LEDGER_STATEMENT_TIMEOUT_MS ): Promise { return withDatabaseReadRetry( (remainingMs) => { const attemptStartedAt = Date.now() return executor.transaction( async (tx) => { - const timeoutMs = Math.floor( - (remainingMs ?? USAGE_LEDGER_STATEMENT_TIMEOUT_MS) - (Date.now() - attemptStartedAt) - ) + const timeoutMs = Math.floor((remainingMs ?? budgetMs) - (Date.now() - attemptStartedAt)) if (timeoutMs <= 0) throw new DatabaseReadDeadlineError() await tx.execute(sql.raw(`SET LOCAL statement_timeout = '${timeoutMs}ms'`)) return read(tx) @@ -33,6 +33,6 @@ export function readLedgerBounded( { accessMode: 'read only' } ) }, - { label: 'usage ledger', maxAttempts: 2, maxElapsedMs: USAGE_LEDGER_STATEMENT_TIMEOUT_MS } + { label: 'usage ledger', maxAttempts: 2, maxElapsedMs: budgetMs } ) } diff --git a/scripts/check-explicit-any.baseline.json b/scripts/check-explicit-any.baseline.json index 85770794ce5..9825662c537 100644 --- a/scripts/check-explicit-any.baseline.json +++ b/scripts/check-explicit-any.baseline.json @@ -1353,7 +1353,6 @@ "apps/sim/hooks/use-permission-config.test.tsx": 10, "apps/sim/hooks/use-permission-config.ts": 1, "apps/sim/hooks/use-voice-input.test.tsx": 1, - "apps/sim/lib/admin/dashboard.ts": 2, "apps/sim/lib/api/contracts/mothership-assistant-tools.test.ts": 2, "apps/sim/lib/api/contracts/mothership-management-tools.test.ts": 1, "apps/sim/lib/api/contracts/v2/tables.test.ts": 1,