diff --git a/apps/sim/lib/admin/dashboard-organizations.test.ts b/apps/sim/lib/admin/dashboard-organizations.test.ts index bb25275d0c1..2b6e74f0892 100644 --- a/apps/sim/lib/admin/dashboard-organizations.test.ts +++ b/apps/sim/lib/admin/dashboard-organizations.test.ts @@ -232,7 +232,8 @@ describe('listDashboardOrganizations', () => { metadata: { invoiceAmountCents: 10_000, seats: 1 }, }, ]) - queueTableRows(usageLog, [{ organizationId: 'org-1', cost: '2.5', workflowRuns: 3 }]) + queueTableRows(usageLog, [{ entityId: 'org-1', cost: '2.5' }]) + queueTableRows(usageLog, [{ entityId: 'org-1', workflowRuns: 3 }]) const result = await listDashboardOrganizations({ search: '', limit: 50, offset: 0 }) @@ -273,7 +274,8 @@ describe('listDashboardOrganizations', () => { }, }, ]) - queueTableRows(usageLog, [{ organizationId: 'org-1', cost: '2.5', workflowRuns: 3 }]) + 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 }) diff --git a/apps/sim/lib/admin/dashboard-usage.integration.ts b/apps/sim/lib/admin/dashboard-usage.integration.ts new file mode 100644 index 00000000000..08eab5903af --- /dev/null +++ b/apps/sim/lib/admin/dashboard-usage.integration.ts @@ -0,0 +1,176 @@ +import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { generateId } from '@sim/utils/id' +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 databaseUrl = readTestDatabaseUrl() +vi.mock('@sim/db', () => ({ db: { select } })) + +import { + type DashboardLedgerWindow, + readDashboardHistoricalActorUsage, + readDashboardLedgerUsage, + readDashboardLedgerUsageByUser, +} from '@/lib/admin/dashboard-usage' +import type { ResolvedUsagePeriod } from '@/lib/billing/core/reporting-period' + +const stampedPeriod: ResolvedUsagePeriod = { + start: new Date('2026-08-01T00:00:00.000Z'), + end: new Date('2026-09-01T00:00:00.000Z'), + source: 'stripe', + anchorDate: null, + interval: 'month', +} +const reportingPeriod: ResolvedUsagePeriod = { + start: new Date('2026-08-15T00:00:00.000Z'), + end: new Date('2026-09-15T00:00:00.000Z'), + source: 'reporting', + 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'` + +const schemaName = `dashboard_usage_${generateId().replaceAll('-', '')}` +const connection = postgres(databaseUrl, { + max: 1, + prepare: false, + connection: { search_path: schemaName }, + onnotice: () => undefined, +}) + +beforeAll(async () => { + await connection.unsafe(`CREATE SCHEMA "${schemaName}"`) + await connection.unsafe(` + CREATE TABLE usage_log ( + id text PRIMARY KEY DEFAULT gen_random_uuid()::text, user_id text NOT NULL, + category text NOT NULL DEFAULT 'model', source text NOT NULL, cost numeric NOT NULL, + billing_entity_type text, billing_entity_id text, billing_period_start timestamp, + billing_period_end timestamp, execution_id text, created_at timestamp NOT NULL + ); + CREATE TABLE member (id text PRIMARY KEY, organization_id text, user_id text); + CREATE TABLE workspace (id text PRIMARY KEY, organization_id text); + CREATE TABLE permissions (id text PRIMARY KEY, entity_type text, entity_id text, user_id text); + + INSERT INTO member VALUES ('m1', 'org-a', 'u1'); + INSERT INTO workspace VALUES ('ws-a', 'org-a'), ('ws-other', 'org-other'); + INSERT INTO permissions VALUES ('p2', 'workspace', 'ws-a', 'u2'), ('p5', 'workspace', 'ws-other', 'u5'); + + INSERT INTO usage_log (user_id, category, source, cost, billing_entity_type, billing_entity_id, + billing_period_start, billing_period_end, execution_id, created_at) VALUES + -- org-a, stamped period: e1 writes one row per charged line and boundary — one run. + ('u1', 'model', 'workflow', 0.5, 'organization', 'org-a', ${P}, 'e1', '2026-08-02'), + ('u1', 'tool', 'workflow', 0.25, 'organization', 'org-a', ${P}, 'e1', '2026-08-02'), + ('u1', 'fixed', 'workflow', 0.01, 'organization', 'org-a', ${P}, 'e1', '2026-08-02'), + -- resumed after the period ended, still stamped with the period it started in + ('u1', 'model', 'workflow', 0.1, 'organization', 'org-a', ${P}, 'e1', '2026-09-02'), + -- a free run is still a run + ('u2', 'model_unbilled', 'workflow', 0, 'organization', 'org-a', ${P}, 'e2', '2026-08-03'), + -- one run with rows from two actors + ('u1', 'model', 'workflow', 0.2, 'organization', 'org-a', ${P}, 'e4', '2026-08-04'), + ('u2', 'model', 'workflow', 0.3, 'organization', 'org-a', ${P}, 'e4', '2026-08-04'), + -- spend that is not a workflow run + ('u1', 'model', 'copilot', 0.4, 'organization', 'org-a', ${P}, NULL, '2026-08-05'), + ('u1', 'model', 'workflow-test', 0.05, 'organization', 'org-a', ${P}, 'e9', '2026-08-05'), + ('u2', 'fixed', 'workflow', 0.07, 'organization', 'org-a', ${P}, NULL, '2026-08-05'), + -- the previous stamped period, though created inside this one + ('u1', 'model', 'workflow', 8, 'organization', 'org-a', ${PREVIOUS}, 'e3', '2026-08-01'), + -- billed to someone else + ('u1', 'model', 'workflow', 9, 'user', 'org-a', ${P}, 'e3b', '2026-08-06'), + ('u1', 'model', 'workflow', 9, NULL, NULL, NULL, NULL, 'e3c', '2026-08-06'), + -- actors who are no longer members or collaborators of org-a + ('u3', 'model', 'workflow', 0.9, 'organization', 'org-a', ${P}, 'e10', '2026-08-07'), + ('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'), + + -- org-b, reporting period: matched on created_at, start inclusive and end exclusive, + -- whatever the rows are stamped with. + ('u6', 'model', 'workflow', 1, 'organization', 'org-b', ${P}, 'e5', '2026-08-15 00:00:00'), + ('u6', 'model', 'workflow', 0.5, 'organization', 'org-b', ${PREVIOUS}, 'e7', '2026-08-20'), + ('u6', 'model', 'workflow', 2, 'organization', 'org-b', ${P}, 'e6', '2026-09-15 00:00:00'), + ('u6', 'model', 'workflow', 4, 'organization', 'org-b', ${P}, 'e6b', '2026-08-14 23:59:59.999'), + -- e1 also billed a line to org-b: each entity counts the run + ('u6', 'model', 'workflow', 0.25, 'organization', 'org-b', ${P}, 'e1', '2026-08-16'), + + -- a personal payer + ('p1', 'model', 'workflow', 0.3, 'user', 'p1', ${P}, 'e8', '2026-08-08'), + ('p1', 'tool', 'workflow', 0.3, 'user', 'p1', ${P}, 'e8', '2026-08-08'); + `) + const database = drizzle(connection) + select.mockImplementation((fields) => database.select(fields)) +}) + +afterAll(async () => { + await connection.unsafe(`DROP SCHEMA "${schemaName}" CASCADE`) + await connection.end() +}) + +describe('dashboard ledger usage', () => { + it('sums spend and counts distinct workflow runs per entity over each period', async () => { + const usage = await readDashboardLedgerUsage([ + 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 }, + }) + }) + + it('reads an entity once however many times it is asked for', async () => { + const usage = await readDashboardLedgerUsage([ + organizationWindow('org-a', stampedPeriod), + organizationWindow('org-a', stampedPeriod), + ]) + expect(usage.get('org-a')).toEqual({ cost: 3.38, workflowRuns: 5 }) + }) + + it('splits spend and runs by acting user, counting a shared run for each actor', async () => { + const usage = await readDashboardLedgerUsageByUser([ + 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 }, + }) + }) + + 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) + }) + + 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, + }) + }) +}) diff --git a/apps/sim/lib/admin/dashboard-usage.ts b/apps/sim/lib/admin/dashboard-usage.ts new file mode 100644 index 00000000000..ac37df8722a --- /dev/null +++ b/apps/sim/lib/admin/dashboard-usage.ts @@ -0,0 +1,215 @@ +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 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' + +/** + * 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. 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. + * + * 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 + * replay delay is cancelled by a recovery conflict, which the largest entities' periods + * can reach. + * + * A workflow run is a distinct `execution_id` among the entity's `source = 'workflow'` + * rows in the period: an execution writes one row per charged line and boundary, so rows + * overcount runs, and only the ledger says which entity a run was billed to. + */ + +export interface DashboardLedgerWindow { + entity: BillingEntity + period: ResolvedUsagePeriod +} + +export interface DashboardLedgerUsage { + cost: number + workflowRuns: number +} + +const WORKFLOW_SOURCE = sql`${usageLog.source} = 'workflow'` + +function windowScope({ entity, period }: DashboardLedgerWindow): SQL[] { + return buildUsageAnalyticsScope(entity, { kind: 'period', period }) +} + +/** Distinct entities only: an arm per repeated entity would count its rows twice. */ +function distinctWindows(windows: DashboardLedgerWindow[]): DashboardLedgerWindow[] { + return [ + ...new Map( + windows.map((window) => [`${window.entity.type}:${window.entity.id}`, window]) + ).values(), + ] +} + +/** + * 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. + */ +async function readLedgerRows(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) => + db + .select({ + entityId: usageLog.billingEntityId, + userId, + cost: sql`coalesce(sum(${usageLog.cost}), 0)`, + }) + .from(usageLog) + .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 +} + +/** Spend and workflow runs per entity id. An entity with no ledger rows is absent. */ +export async function readDashboardLedgerUsage( + windows: DashboardLedgerWindow[] +): 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 + } + 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( + windows: DashboardLedgerWindow[], + userIds?: string[] +): 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) + 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 + } + } + return result +} + +const historicalUsageMember = alias(member, 'historical_usage_member') +const historicalUsagePermission = alias(permissions, 'historical_usage_permission') +const historicalUsageWorkspace = alias(workspace, 'historical_usage_workspace') + +/** + * 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. + */ +export async function readDashboardHistoricalActorUsage( + 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) + ) + ) + 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), + ] + + 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, + } +} diff --git a/apps/sim/lib/admin/dashboard.ts b/apps/sim/lib/admin/dashboard.ts index 692aa874028..bdb6add9ba2 100644 --- a/apps/sim/lib/admin/dashboard.ts +++ b/apps/sim/lib/admin/dashboard.ts @@ -7,29 +7,18 @@ import { outboxEvent, permissions, subscription, - usageLog, user, userStats, workspace, } from '@sim/db/schema' 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 { - and, - count, - countDistinct, - desc, - eq, - gte, - ilike, - inArray, - isNull, - lt, - notExists, - or, - sql, -} from 'drizzle-orm' -import { alias } from 'drizzle-orm/pg-core' + readDashboardHistoricalActorUsage, + readDashboardLedgerUsage, + readDashboardLedgerUsageByUser, +} from '@/lib/admin/dashboard-usage' import { getOrganizationUsageLimitFallbackDollars, getTeamOrganizationEconomics, @@ -216,148 +205,46 @@ async function getDashboardOrganizationUsage( const includeUserBreakdown = options.includeUserBreakdown ?? true if (options.userIds?.length === 0) return result - const ledgerPeriodWhere = or( - ...contexts.map((context) => - and( - eq(usageLog.billingEntityType, 'organization'), - eq(usageLog.billingEntityId, context.organizationId), - ...(context.period.source === 'reporting' - ? [ - gte(usageLog.createdAt, context.period.start), - lt(usageLog.createdAt, context.period.end), - ] - : [ - eq(usageLog.billingPeriodStart, context.period.start), - eq(usageLog.billingPeriodEnd, context.period.end), - ]) - ) - ) - ) + const windows = contexts.map(({ organizationId, period }) => ({ + entity: { type: 'organization' as const, id: organizationId }, + period, + })) if (!includeUserBreakdown) { - const ledgerTotals = await db - .select({ - organizationId: usageLog.billingEntityId, - cost: sql`coalesce(sum(${usageLog.cost}), 0)`, - workflowRuns: - sql`count(distinct ${usageLog.executionId}) filter (where ${usageLog.source} = 'workflow')`.mapWith( - Number - ), - }) - .from(usageLog) - .where(ledgerPeriodWhere) - .groupBy(usageLog.billingEntityId) - for (const row of ledgerTotals) { - if (!row.organizationId) continue - const usage = result.get(row.organizationId) + for (const [organizationId, ledger] of await readDashboardLedgerUsage(windows)) { + const usage = result.get(organizationId) if (usage) { - usage.total += Number(row.cost) - usage.workflowRuns += row.workflowRuns + usage.total += ledger.cost + usage.workflowRuns += ledger.workflowRuns } } - return result } - const ledgerRows = await db - .select({ - organizationId: usageLog.billingEntityId, - userId: usageLog.userId, - cost: sql`coalesce(sum(${usageLog.cost}), 0)`, - workflowRuns: - sql`count(distinct ${usageLog.executionId}) filter (where ${usageLog.source} = 'workflow')`.mapWith( - Number - ), - }) - .from(usageLog) - .where( - options.userIds - ? and(ledgerPeriodWhere, inArray(usageLog.userId, options.userIds)) - : ledgerPeriodWhere - ) - .groupBy(usageLog.billingEntityId, usageLog.userId) - - for (const row of ledgerRows) { - if (!row.organizationId) continue - const usage = result.get(row.organizationId) + const ledgerByUser = await readDashboardLedgerUsageByUser(windows, options.userIds) + for (const [organizationId, users] of ledgerByUser) { + const usage = result.get(organizationId) if (!usage) continue - const amount = Number(row.cost) - usage.total += amount - usage.byUser.set(row.userId, (usage.byUser.get(row.userId) ?? 0) + amount) - usage.workflowRuns += row.workflowRuns - usage.workflowRunsByUser.set( - row.userId, - (usage.workflowRunsByUser.get(row.userId) ?? 0) + row.workflowRuns - ) + 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 } -const historicalUsageMember = alias(member, 'historical_usage_member') -const historicalUsagePermission = alias(permissions, 'historical_usage_permission') -const historicalUsageWorkspace = alias(workspace, 'historical_usage_workspace') - async function getHistoricalActorUsage( organizationId: string, period: ResolvedUsagePeriod ): Promise<{ usedDollars: number; usedCredits: 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) - ) - ) - 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 [row] = await db - .select({ - usedDollars: sql`coalesce(sum(${usageLog.cost}), 0)`, - workflowRuns: - sql`count(distinct ${usageLog.executionId}) filter (where ${usageLog.source} = 'workflow')`.mapWith( - Number - ), - actorCount: countDistinct(usageLog.userId), - }) - .from(usageLog) - .where( - and( - eq(usageLog.billingEntityType, 'organization'), - eq(usageLog.billingEntityId, organizationId), - ...(period.source === 'reporting' - ? [gte(usageLog.createdAt, period.start), lt(usageLog.createdAt, period.end)] - : [ - eq(usageLog.billingPeriodStart, period.start), - eq(usageLog.billingPeriodEnd, period.end), - ]), - notExists(currentMember), - notExists(currentCollaborator) - ) - ) - const usedDollars = Number(row?.usedDollars ?? 0) - return { - usedDollars, - usedCredits: dollarsToCredits(usedDollars), - workflowRuns: row?.workflowRuns ?? 0, - actorCount: row?.actorCount ?? 0, - } + const usage = await readDashboardHistoricalActorUsage(organizationId, period) + return { ...usage, usedCredits: dollarsToCredits(usage.usedDollars) } } export function toDashboardProvisioning(view: EnterpriseProvisioningView) { @@ -573,49 +460,11 @@ export async function listDashboardUsers({ search, limit, offset }: PaginationIn const personalSubscriptionMap = new Map( personalSubscriptions.map((row) => [row.referenceId, row]) ) - const personalPeriods = new Map( - personalUserIds.map((userId) => [ - userId, - resolveDashboardUsagePeriod(personalSubscriptionMap.get(userId) ?? null), - ]) - ) - const personalLedgerRows = - personalUserIds.length === 0 - ? [] - : await db - .select({ - userId: usageLog.billingEntityId, - cost: sql`coalesce(sum(${usageLog.cost}), 0)`, - workflowRuns: - sql`count(distinct ${usageLog.executionId}) filter (where ${usageLog.source} = 'workflow')`.mapWith( - Number - ), - }) - .from(usageLog) - .where( - or( - ...personalUserIds.map((userId) => { - const period = personalPeriods.get(userId) as ResolvedUsagePeriod - return and( - eq(usageLog.billingEntityType, 'user'), - eq(usageLog.billingEntityId, userId), - ...(period.source === 'reporting' - ? [gte(usageLog.createdAt, period.start), lt(usageLog.createdAt, period.end)] - : [ - eq(usageLog.billingPeriodStart, period.start), - eq(usageLog.billingPeriodEnd, period.end), - ]) - ) - }) - ) - ) - .groupBy(usageLog.billingEntityId) - const personalUsage = new Map( - personalLedgerRows.flatMap((row) => - row.userId - ? ([[row.userId, { dollars: Number(row.cost), workflowRuns: row.workflowRuns }]] as const) - : [] - ) + const personalUsage = await readDashboardLedgerUsage( + personalUserIds.map((userId) => ({ + entity: { type: 'user' as const, id: userId }, + period: resolveDashboardUsagePeriod(personalSubscriptionMap.get(userId) ?? null), + })) ) return { @@ -625,7 +474,7 @@ export async function listDashboardUsers({ search, limit, offset }: PaginationIn : undefined const usageDollars = row.organizationId ? (organization?.byUser.get(row.id) ?? 0) - : (personalUsage.get(row.id)?.dollars ?? 0) + : (personalUsage.get(row.id)?.cost ?? 0) return { id: row.id, name: row.name,