mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
DottaandPaperclip
0eaabe5c0b
Settle paid operations through operation-scoped transactions. Preserve exact ledger evidence and replay while moving aggregate budget decisions to durable deferred work. Co-Authored-By: Paperclip <noreply@paperclip.ing>
687 lines
36 KiB
TypeScript
687 lines
36 KiB
TypeScript
import { receiptFingerprint } from "./receipt-fingerprint.js";
|
|
import { agentAvatarUrl, resolveAgentAppearance, createCostEventSchema, createServiceCostEventSchema, normalizeCents, isUuidLike, type MoneyInput, type CostByUserReport } from "@paperclipai/shared";
|
|
import { and, desc, eq, gte, inArray, isNotNull, isNull, lt, lte, ne, sql } from "drizzle-orm";
|
|
import { alias } from "drizzle-orm/pg-core";
|
|
import type { Db } from "@paperclipai/db";
|
|
import { activityLog, agents, authUsers, companies, companyMemberships, costEvents, heartbeatRuns, issues, projects, goals, aiSubscriptions } from "@paperclipai/db";
|
|
import { notFound, unprocessable, conflict } from "../errors.js";
|
|
import type { BudgetServiceHooks } from "./budgets.js";
|
|
import { logActivity, type LogActivityInput, type ActivityPublication } from "./activity-log.js";
|
|
import { withAccountingReadSnapshot, withOperationAccountingTransaction } from "./accounting-transaction.js";
|
|
import { queueBudgetEvaluation } from "./budget-evaluation-work.js";
|
|
import { visibleIssueCondition } from "./issue-visibility.js";
|
|
|
|
import { resolveCostDateRange, type CostDateRange } from "./cost-date-range.js";
|
|
export type { CostDateRange } from "./cost-date-range.js";
|
|
|
|
const METERED_BILLING_TYPE = "metered_api";
|
|
const SUBSCRIPTION_BILLING_TYPES = ["subscription_included", "subscription_overage"] as const;
|
|
|
|
// Pre-receipt Codex events stored cache reads inside input; Pi/OpenCode used
|
|
// provider-qualified model IDs and already excluded cache reads. Normalize each
|
|
// historical row before aggregation so mixed old/new groups share the current
|
|
// exclusive-input contract. Preserve the original ledger and monetary amounts.
|
|
export const ordinaryInputTokens = sql<number>`case
|
|
when ${costEvents.receiptHash} is null and ${costEvents.provider} = 'openai'
|
|
and ${costEvents.model} not like 'openai/_%'
|
|
then greatest(0, ${costEvents.inputTokens} - ${costEvents.cachedInputTokens})
|
|
else ${costEvents.inputTokens} end`;
|
|
|
|
function sumAsNumber(column: typeof costEvents.costCents | typeof costEvents.inputTokens | typeof costEvents.cachedInputTokens | typeof costEvents.outputTokens | ReturnType<typeof sql>) {
|
|
return sql<number>`coalesce(sum(${column}), 0)::double precision`;
|
|
}
|
|
|
|
function currentUtcMonthWindow(now = new Date()) {
|
|
const year = now.getUTCFullYear();
|
|
const month = now.getUTCMonth();
|
|
return {
|
|
start: new Date(Date.UTC(year, month, 1, 0, 0, 0, 0)),
|
|
end: new Date(Date.UTC(year, month + 1, 1, 0, 0, 0, 0)),
|
|
};
|
|
}
|
|
|
|
export async function getMonthlySpendTotal(
|
|
db: Db,
|
|
scope: { companyId: string; agentId?: string | null },
|
|
) {
|
|
const { start, end } = currentUtcMonthWindow();
|
|
const conditions = [
|
|
eq(costEvents.companyId, scope.companyId),
|
|
gte(costEvents.occurredAt, start),
|
|
lt(costEvents.occurredAt, end),
|
|
];
|
|
if (scope.agentId) {
|
|
conditions.push(eq(costEvents.agentId, scope.agentId));
|
|
}
|
|
const [row] = await db
|
|
.select({
|
|
total: sumAsNumber(costEvents.costCents),
|
|
})
|
|
.from(costEvents)
|
|
.where(and(...conditions));
|
|
return Number(row?.total ?? 0);
|
|
}
|
|
|
|
export async function createCostEventInTransaction(db: Db, companyId: string, data: Omit<typeof costEvents.$inferInsert, "companyId" | "receiptHash" | "costCents"> & { costCents: MoneyInput }, publications: ActivityPublication[] = [], actor?: Pick<LogActivityInput, "actorType" | "actorId" | "agentId">,
|
|
options: { deferBudgetEvaluation?: boolean } = {}) {
|
|
const parsed = ((data.usageKind === "decision" || data.usageKind === "fast_response" || data.usageKind === "prompt_injection") ? createServiceCostEventSchema : createCostEventSchema).safeParse({ ...data, occurredAt: data.occurredAt.toISOString() });
|
|
if (!parsed.success) throw unprocessable("Invalid cost receipt", parsed.error.flatten());
|
|
const values = {
|
|
...parsed.data,
|
|
issueId: parsed.data.issueId ?? null,
|
|
projectId: parsed.data.projectId ?? null,
|
|
goalId: parsed.data.goalId ?? null,
|
|
heartbeatRunId: parsed.data.heartbeatRunId ?? null,
|
|
billingCode: parsed.data.billingCode ?? null,
|
|
idempotencyKey: parsed.data.idempotencyKey ?? null,
|
|
costCents: normalizeCents(parsed.data.costCents),
|
|
occurredAt: new Date(parsed.data.occurredAt),
|
|
};
|
|
const receiptHash = receiptFingerprint(values);
|
|
// Older receipts hashed the numeric API value. Retain replay compatibility
|
|
// only when the decimal value survives that representation exactly.
|
|
let legacyHash: string | null = null;
|
|
try {
|
|
if (normalizeCents(Number(values.costCents)) === values.costCents) legacyHash = receiptFingerprint({ ...values, costCents: Number(values.costCents) });
|
|
} catch { /* The legacy number may round beyond the decimal storage limit. */ }
|
|
if (values.idempotencyKey) {
|
|
const [existing] = await db.select().from(costEvents).where(and(
|
|
eq(costEvents.companyId, companyId), eq(costEvents.idempotencyKey, values.idempotencyKey),
|
|
));
|
|
if (existing) {
|
|
if (existing.receiptHash !== receiptHash && existing.receiptHash !== legacyHash) throw conflict("Idempotency key already used for a different cost receipt");
|
|
const [exact] = await db.select({ costCents: sql<string>`${costEvents.costCents}::text` }).from(costEvents).where(eq(costEvents.id, existing.id));
|
|
return { ...existing, costCentsExact: normalizeCents(exact.costCents) };
|
|
}
|
|
}
|
|
const links = [
|
|
[agents, values.agentId, "Agent"], [issues, values.issueId, "Issue"],
|
|
[projects, values.projectId, "Project"], [goals, values.goalId, "Goal"],
|
|
[heartbeatRuns, values.heartbeatRunId, "Heartbeat run"],
|
|
] as const;
|
|
for (const [table, id, label] of links) {
|
|
if (!id) continue;
|
|
const [row] = await db.select({ companyId: table.companyId }).from(table).where(eq(table.id, id));
|
|
if (!row) throw notFound(`${label} not found`);
|
|
if (row.companyId !== companyId) throw unprocessable(`${label} does not belong to company`);
|
|
}
|
|
let subscriptionId: string | null = null;
|
|
if (values.heartbeatRunId) {
|
|
const [run] = await db.select({ agentId: heartbeatRuns.agentId, context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(eq(heartbeatRuns.id, values.heartbeatRunId));
|
|
if (run.agentId !== values.agentId) throw unprocessable("Heartbeat run does not belong to agent");
|
|
const attribution = run.context?.aiConnection as { subscriptionId?: unknown; provider?: unknown } | undefined;
|
|
if (SUBSCRIPTION_BILLING_TYPES.includes(values.billingType as typeof SUBSCRIPTION_BILLING_TYPES[number]) && attribution?.provider === values.provider && typeof attribution.subscriptionId === "string" && isUuidLike(attribution.subscriptionId)) {
|
|
const [account] = await db.select({ id: aiSubscriptions.id }).from(aiSubscriptions).where(and(eq(aiSubscriptions.id, attribution.subscriptionId), eq(aiSubscriptions.companyId, companyId), eq(aiSubscriptions.provider, values.provider)));
|
|
subscriptionId = account?.id ?? null;
|
|
}
|
|
}
|
|
// Server-derived attribution stays outside the immutable receipt fingerprint;
|
|
// replaying a receipt must not depend on later account metadata changes.
|
|
const [inserted] = await db.insert(costEvents).values({ ...values, subscriptionId, costCents: sql`${values.costCents}::numeric`, reportedCostCents: values.costCents, id: data.id, companyId, receiptHash })
|
|
.onConflictDoNothing().returning();
|
|
if (!inserted) {
|
|
const [winner] = values.idempotencyKey ? await db.select().from(costEvents).where(and(
|
|
eq(costEvents.companyId, companyId), eq(costEvents.idempotencyKey, values.idempotencyKey))) : [];
|
|
if (!winner) throw Object.assign(new Error("Concurrent cost receipt requires a fresh snapshot"), { code: "40001" });
|
|
if (winner.receiptHash !== receiptHash && winner.receiptHash !== legacyHash)
|
|
throw conflict("Idempotency key already used for a different cost receipt");
|
|
const [exact] = await db.select({ costCents: sql<string>`${costEvents.costCents}::text` }).from(costEvents).where(eq(costEvents.id, winner.id));
|
|
return { ...winner, costCentsExact: normalizeCents(exact.costCents) };
|
|
}
|
|
const event = inserted;
|
|
if (!options.deferBudgetEvaluation) await queueBudgetEvaluation(db, { companyId, agentId: event.agentId, projectId: event.projectId });
|
|
await logActivity(db, {
|
|
companyId, actorType: actor?.actorType ?? "system", actorId: actor?.actorId ?? "cost_accounting", agentId: actor?.agentId ?? event.agentId,
|
|
...((event.usageKind === "decision" || event.usageKind === "fast_response" || event.usageKind === "prompt_injection") ? { responsibleUserIdOverride: event.responsibleUserId } : {}),
|
|
runId: event.heartbeatRunId, action: "cost.reported", entityType: "cost_event", entityId: event.id,
|
|
details: { costCents: event.costCents, costCentsExact: values.costCents, model: event.model, costStatus: event.costStatus },
|
|
}, publications);
|
|
return { ...event, costCentsExact: values.costCents };
|
|
}
|
|
|
|
export function costService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
|
|
return {
|
|
createEvent: async (companyId: string, data: Omit<typeof costEvents.$inferInsert, "companyId" | "receiptHash" | "costCents"> & { costCents: MoneyInput }, actor?: Pick<LogActivityInput, "actorType" | "actorId" | "agentId">) => {
|
|
if (data.idempotencyKey?.startsWith("heartbeat:")) throw unprocessable("The heartbeat idempotency namespace is reserved for run receipts");
|
|
const event = await withOperationAccountingTransaction(db, companyId, (tx, publications) => createCostEventInTransaction(tx, companyId, data, publications, actor));
|
|
return event;
|
|
},
|
|
|
|
summary: async (companyId: string, range?: CostDateRange) => withAccountingReadSnapshot(db, companyId, async (db) => {
|
|
const company = await db
|
|
.select()
|
|
.from(companies)
|
|
.where(eq(companies.id, companyId))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
if (!company) throw notFound("Company not found");
|
|
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
const [{ total, totalExact, eventCount, unpricedEventCount, estimatedEventCount, unknownBillingEventCount }] = await db
|
|
.select({
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
unknownBillingEventCount: sql<number>`count(*) filter (where ${costEvents.billingType} = 'unknown')::int`,
|
|
unpricedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`,
|
|
total: sumAsNumber(costEvents.costCents),
|
|
totalExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
})
|
|
.from(costEvents)
|
|
.where(and(...conditions));
|
|
|
|
const pendingConditions = [eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.costAccountingPending, true),
|
|
inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])];
|
|
if (range?.from) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) >= ${range.from.toISOString()}::timestamptz`);
|
|
if (range?.to) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) <= ${range.to.toISOString()}::timestamptz`);
|
|
const [pending] = await db.select({ count: sql<number>`count(*)::int` }).from(heartbeatRuns).where(and(...pendingConditions));
|
|
const pendingRunCount = pending?.count ?? 0;
|
|
const spendCents = Number(total);
|
|
const utilization =
|
|
company.budgetMonthlyCents > 0
|
|
? (spendCents / company.budgetMonthlyCents) * 100
|
|
: 0;
|
|
|
|
return {
|
|
companyId,
|
|
eventCount,
|
|
unpricedEventCount,
|
|
unknownBillingEventCount,
|
|
estimatedEventCount,
|
|
pendingRunCount,
|
|
pricingComplete: unpricedEventCount === 0 && pendingRunCount === 0,
|
|
spendCents,
|
|
spendCentsExact: normalizeCents(totalExact),
|
|
budgetCents: company.budgetMonthlyCents,
|
|
utilizationPercent: Number(utilization.toFixed(2)),
|
|
};
|
|
}),
|
|
|
|
issueTreeSummary: async (
|
|
companyId: string,
|
|
issueId: string,
|
|
options: { excludeRoot?: boolean } = {},
|
|
) => {
|
|
// Callers must resolve and authorize a visible root issue before invoking this.
|
|
// The route does that so zero counts are not mistaken for a missing root.
|
|
const childIssues = alias(issues, "child");
|
|
|
|
// The seed of the recursive CTE: when excludeRoot is true, start from
|
|
// the direct children so the root issue itself is not counted.
|
|
const cteSeed = options.excludeRoot
|
|
? sql`
|
|
SELECT ${issues.id}
|
|
FROM ${issues}
|
|
WHERE ${issues.companyId} = ${companyId}
|
|
AND ${issues.parentId} = ${issueId}
|
|
AND ${issues.hiddenAt} IS NULL
|
|
AND ${issues.harnessKind} IS NULL
|
|
`
|
|
: sql`
|
|
SELECT ${issues.id}
|
|
FROM ${issues}
|
|
WHERE ${issues.companyId} = ${companyId}
|
|
AND ${issues.id} = ${issueId}
|
|
AND ${issues.hiddenAt} IS NULL
|
|
AND ${issues.harnessKind} IS NULL
|
|
`;
|
|
|
|
const cteSeedText = options.excludeRoot
|
|
? sql`
|
|
SELECT (${issues.id})::text AS id
|
|
FROM ${issues}
|
|
WHERE ${issues.companyId} = ${companyId}
|
|
AND ${issues.parentId} = ${issueId}
|
|
AND ${issues.hiddenAt} IS NULL
|
|
AND ${issues.harnessKind} IS NULL
|
|
`
|
|
: sql`
|
|
SELECT (${issues.id})::text AS id
|
|
FROM ${issues}
|
|
WHERE ${issues.companyId} = ${companyId}
|
|
AND ${issues.id} = ${issueId}
|
|
AND ${issues.hiddenAt} IS NULL
|
|
AND ${issues.harnessKind} IS NULL
|
|
`;
|
|
|
|
const issueTreeCondition = sql<boolean>`
|
|
${issues.id} IN (
|
|
WITH RECURSIVE issue_tree(id) AS (
|
|
${cteSeed}
|
|
UNION ALL
|
|
SELECT ${childIssues.id}
|
|
FROM ${issues} ${childIssues}
|
|
JOIN issue_tree ON ${childIssues.parentId} = issue_tree.id
|
|
WHERE ${childIssues.companyId} = ${companyId}
|
|
AND ${childIssues.hiddenAt} IS NULL
|
|
AND ${childIssues.harnessKind} IS NULL
|
|
)
|
|
SELECT id FROM issue_tree
|
|
)
|
|
`;
|
|
|
|
const runSummarySql = sql`
|
|
WITH RECURSIVE issue_tree(id) AS (
|
|
${cteSeedText}
|
|
UNION ALL
|
|
SELECT (${childIssues.id})::text
|
|
FROM ${issues} ${childIssues}
|
|
JOIN issue_tree ON (${childIssues.parentId})::text = issue_tree.id
|
|
WHERE ${childIssues.companyId} = ${companyId}
|
|
AND ${childIssues.hiddenAt} IS NULL
|
|
AND ${childIssues.harnessKind} IS NULL
|
|
)
|
|
SELECT
|
|
count(distinct ${heartbeatRuns.id})::int AS "runCount",
|
|
coalesce(sum(extract(epoch from (coalesce(${heartbeatRuns.finishedAt}, now()) - ${heartbeatRuns.startedAt})) * 1000), 0)::double precision AS "runtimeMs"
|
|
FROM ${heartbeatRuns}
|
|
WHERE ${heartbeatRuns.companyId} = ${companyId}
|
|
AND ${heartbeatRuns.startedAt} IS NOT NULL
|
|
AND (
|
|
${heartbeatRuns.contextSnapshot} ->> 'issueId' IN (SELECT id FROM issue_tree)
|
|
OR EXISTS (
|
|
SELECT 1
|
|
FROM ${activityLog}
|
|
JOIN issue_tree ON ${activityLog.entityId} = issue_tree.id
|
|
WHERE ${activityLog.companyId} = ${companyId}
|
|
AND ${activityLog.entityType} = 'issue'
|
|
AND ${activityLog.runId} = ${heartbeatRuns.id}
|
|
)
|
|
)
|
|
`;
|
|
|
|
// Run cost-event aggregation and run-duration aggregation in parallel.
|
|
// They're separate queries because cost_events fan out per-event and
|
|
// joining heartbeat_runs through them would double-count run durations.
|
|
const [costRowResult, runRowResult] = await Promise.all([
|
|
db
|
|
.select({
|
|
issueCount: sql<number>`count(distinct ${issues.id})::int`,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
})
|
|
.from(issues)
|
|
.leftJoin(
|
|
costEvents,
|
|
and(
|
|
eq(costEvents.companyId, companyId),
|
|
eq(costEvents.issueId, issues.id),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(issues.companyId, companyId),
|
|
visibleIssueCondition(),
|
|
issueTreeCondition,
|
|
),
|
|
),
|
|
db.execute(runSummarySql),
|
|
]);
|
|
|
|
const costRow = costRowResult[0];
|
|
const runRow = Array.isArray(runRowResult)
|
|
? (runRowResult[0] as { runCount?: number | string | null; runtimeMs?: number | string | null } | undefined)
|
|
: undefined;
|
|
|
|
return {
|
|
issueId,
|
|
issueCount: Number(costRow?.issueCount ?? 0),
|
|
includeDescendants: true,
|
|
costCents: Number(costRow?.costCents ?? 0),
|
|
costCentsExact: normalizeCents(costRow?.costCentsExact ?? 0),
|
|
inputTokens: Number(costRow?.inputTokens ?? 0),
|
|
cachedInputTokens: Number(costRow?.cachedInputTokens ?? 0),
|
|
outputTokens: Number(costRow?.outputTokens ?? 0),
|
|
runCount: Number(runRow?.runCount ?? 0),
|
|
runtimeMs: Number(runRow?.runtimeMs ?? 0),
|
|
};
|
|
},
|
|
|
|
byUser: (companyId: string, range?: CostDateRange): Promise<CostByUserReport> => withAccountingReadSnapshot(db, companyId, async tx => {
|
|
// Keep membership, identity and spend in one non-blocking read snapshot.
|
|
const members = await tx.select({
|
|
userId: companyMemberships.principalId,
|
|
status: companyMemberships.status,
|
|
userName: authUsers.name,
|
|
userImage: authUsers.image,
|
|
}).from(companyMemberships)
|
|
.leftJoin(authUsers, eq(authUsers.id, companyMemberships.principalId))
|
|
.where(and(
|
|
eq(companyMemberships.companyId, companyId),
|
|
eq(companyMemberships.principalType, "user"),
|
|
ne(companyMemberships.principalId, "local-board"),
|
|
));
|
|
const activeMembers = members.filter(member => member.status === "active");
|
|
const activeUserCount = activeMembers.length;
|
|
|
|
const bounds = resolveCostDateRange(range);
|
|
const conditions = [eq(costEvents.companyId, companyId)];
|
|
if (bounds.from) conditions.push(gte(costEvents.occurredAt, bounds.from));
|
|
if (bounds.to) conditions.push(lte(costEvents.occurredAt, bounds.to));
|
|
const totals = await tx.select({
|
|
userId: companyMemberships.principalId,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
unpricedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
runCount: sql<number>`count(distinct ${heartbeatRuns.id})::int`,
|
|
}).from(costEvents)
|
|
.leftJoin(heartbeatRuns, and(
|
|
eq(heartbeatRuns.id, costEvents.heartbeatRunId),
|
|
eq(heartbeatRuns.companyId, costEvents.companyId),
|
|
eq(heartbeatRuns.agentId, costEvents.agentId),
|
|
))
|
|
// Never resolve names through a foreign-company run or membership. Retain
|
|
// former members' historical spend; missing attribution remains in the total.
|
|
// The synthetic Board principal is not a person. Any historical charges
|
|
// attached to it remain in Unattributed so the report still reconciles.
|
|
.leftJoin(companyMemberships, and(
|
|
eq(companyMemberships.companyId, companyId),
|
|
eq(companyMemberships.principalType, "user"),
|
|
ne(companyMemberships.principalId, "local-board"),
|
|
eq(companyMemberships.principalId, sql`case when ${costEvents.usageKind} in ('decision', 'fast_response', 'prompt_injection') then ${costEvents.responsibleUserId} else ${heartbeatRuns.responsibleUserId} end`),
|
|
))
|
|
.where(and(...conditions))
|
|
.groupBy(companyMemberships.principalId)
|
|
.orderBy(desc(sql`sum(${costEvents.costCents})`), companyMemberships.principalId);
|
|
const directory = new Map(members.map(member => [member.userId, member]));
|
|
const rows: CostByUserReport["rows"] = totals.map(row => ({
|
|
...row,
|
|
userName: row.userId ? directory.get(row.userId)?.userName ?? "Former user" : null,
|
|
userImage: row.userId ? directory.get(row.userId)?.userImage ?? null : null,
|
|
costCentsExact: normalizeCents(row.costCentsExact),
|
|
}));
|
|
const represented = new Set(rows.map(row => row.userId));
|
|
for (const member of activeMembers.sort((a, b) => (a.userName ?? a.userId).localeCompare(b.userName ?? b.userId))) {
|
|
if (represented.has(member.userId)) continue;
|
|
rows.push({
|
|
userId: member.userId, userName: member.userName ?? "Unknown user", userImage: member.userImage,
|
|
eventCount: 0, estimatedEventCount: 0, unpricedEventCount: 0,
|
|
costCents: 0, costCentsExact: "0.0000000",
|
|
inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, runCount: 0,
|
|
});
|
|
}
|
|
return { activeUserCount, rows };
|
|
}),
|
|
|
|
byAgent: async (companyId: string, range?: CostDateRange) => {
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
const rows = await db
|
|
.select({
|
|
agentId: costEvents.agentId,
|
|
agentName: agents.name,
|
|
agentAppearance: agents.appearance,
|
|
agentStatus: agents.status,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
apiRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionCachedInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionOutputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
|
|
})
|
|
.from(costEvents)
|
|
.leftJoin(agents, eq(costEvents.agentId, agents.id))
|
|
.where(and(...conditions))
|
|
.groupBy(costEvents.agentId, agents.name, agents.appearance, agents.status)
|
|
.orderBy(desc(sumAsNumber(costEvents.costCents)));
|
|
return rows.map(row => {
|
|
const appearance = row.agentId ? resolveAgentAppearance(row.agentAppearance, row.agentId) : null;
|
|
return { ...row, agentAppearance: appearance, avatarUrl: appearance ? agentAvatarUrl(appearance, 512) : undefined };
|
|
});
|
|
},
|
|
|
|
byProvider: async (companyId: string, range?: CostDateRange) => {
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
return db
|
|
.select({
|
|
provider: costEvents.provider,
|
|
biller: costEvents.biller,
|
|
billingType: costEvents.billingType,
|
|
model: costEvents.model,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
unpricedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
apiRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionCachedInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionOutputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
|
|
})
|
|
.from(costEvents)
|
|
.where(and(...conditions))
|
|
.groupBy(costEvents.provider, costEvents.biller, costEvents.billingType, costEvents.model)
|
|
.orderBy(desc(sumAsNumber(costEvents.costCents)));
|
|
},
|
|
|
|
byBiller: async (companyId: string, range?: CostDateRange) => {
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
return db
|
|
.select({
|
|
biller: costEvents.biller,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
unpricedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
apiRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} = ${METERED_BILLING_TYPE} then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionRunCount:
|
|
sql<number>`count(distinct case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.heartbeatRunId} end)::int`,
|
|
subscriptionCachedInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionInputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
|
|
subscriptionOutputTokens:
|
|
sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
|
|
providerCount: sql<number>`count(distinct ${costEvents.provider})::int`,
|
|
modelCount: sql<number>`count(distinct ${costEvents.model})::int`,
|
|
})
|
|
.from(costEvents)
|
|
.where(and(...conditions))
|
|
.groupBy(costEvents.biller)
|
|
.orderBy(desc(sumAsNumber(costEvents.costCents)));
|
|
},
|
|
|
|
/**
|
|
* aggregates cost_events by provider for each of three rolling windows:
|
|
* last 5 hours, last 24 hours, last 7 days.
|
|
* purely internal consumption data, no external rate-limit sources.
|
|
*/
|
|
windowSpend: async (companyId: string) => {
|
|
const windows = [
|
|
{ label: "5h", hours: 5 },
|
|
{ label: "24h", hours: 24 },
|
|
{ label: "7d", hours: 168 },
|
|
] as const;
|
|
|
|
const now = new Date();
|
|
const results = await Promise.all(
|
|
windows.map(async ({ label, hours }) => {
|
|
const since = new Date(now.getTime() - hours * 60 * 60 * 1000);
|
|
const rows = await db
|
|
.select({
|
|
provider: costEvents.provider,
|
|
biller: sql<string>`case when count(distinct ${costEvents.biller}) = 1 then min(${costEvents.biller}) else 'mixed' end`,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
})
|
|
.from(costEvents)
|
|
.where(
|
|
and(
|
|
eq(costEvents.companyId, companyId),
|
|
gte(costEvents.occurredAt, since),
|
|
lte(costEvents.occurredAt, now),
|
|
),
|
|
)
|
|
.groupBy(costEvents.provider)
|
|
.orderBy(desc(sumAsNumber(costEvents.costCents)));
|
|
|
|
return rows.map((row) => ({
|
|
provider: row.provider,
|
|
biller: row.biller,
|
|
window: label as string,
|
|
windowHours: hours,
|
|
costCents: row.costCents,
|
|
costCentsExact: normalizeCents(row.costCentsExact),
|
|
inputTokens: row.inputTokens,
|
|
cachedInputTokens: row.cachedInputTokens,
|
|
outputTokens: row.outputTokens,
|
|
}));
|
|
}),
|
|
);
|
|
|
|
return results.flat();
|
|
},
|
|
|
|
byAgentModel: async (companyId: string, range?: CostDateRange) => {
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
// single query: group by agent + provider + model.
|
|
// the (companyId, agentId, occurredAt) composite index covers this well.
|
|
// order by provider + model for stable db-level ordering; cost-desc sort
|
|
// within each agent's sub-rows is done client-side in the ui memo.
|
|
const rows = await db
|
|
.select({
|
|
agentId: costEvents.agentId,
|
|
agentName: agents.name,
|
|
agentAppearance: agents.appearance,
|
|
provider: costEvents.provider,
|
|
biller: costEvents.biller,
|
|
billingType: costEvents.billingType,
|
|
model: costEvents.model,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
costCents: sumAsNumber(costEvents.costCents),
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
})
|
|
.from(costEvents)
|
|
.leftJoin(agents, eq(costEvents.agentId, agents.id))
|
|
.where(and(...conditions))
|
|
.groupBy(
|
|
costEvents.agentId,
|
|
agents.name,
|
|
agents.appearance,
|
|
costEvents.provider,
|
|
costEvents.biller,
|
|
costEvents.billingType,
|
|
costEvents.model,
|
|
)
|
|
.orderBy(costEvents.provider, costEvents.biller, costEvents.billingType, costEvents.model);
|
|
return rows.map(row => {
|
|
const appearance = row.agentId ? resolveAgentAppearance(row.agentAppearance, row.agentId) : null;
|
|
return { ...row, agentAppearance: appearance, avatarUrl: appearance ? agentAvatarUrl(appearance, 512) : undefined };
|
|
});
|
|
},
|
|
|
|
byProject: async (companyId: string, range?: CostDateRange) => {
|
|
const issueIdAsText = sql<string>`${issues.id}::text`;
|
|
const runProjectLinks = db
|
|
.select({
|
|
runId: activityLog.runId,
|
|
projectId: sql<string>`min(${issues.projectId}::text)::uuid`.as("linked_project_id"),
|
|
})
|
|
.from(activityLog)
|
|
.innerJoin(
|
|
issues,
|
|
and(
|
|
eq(activityLog.entityType, "issue"),
|
|
eq(activityLog.entityId, issueIdAsText),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(activityLog.companyId, companyId),
|
|
eq(issues.companyId, companyId),
|
|
isNotNull(activityLog.runId),
|
|
isNotNull(issues.projectId),
|
|
),
|
|
)
|
|
.groupBy(activityLog.runId)
|
|
.having(sql`count(distinct ${issues.projectId}) = 1`)
|
|
.as("run_project_links");
|
|
|
|
const effectiveProjectId = sql<string | null>`coalesce(${costEvents.projectId}, case when ${costEvents.receiptHash} is null then ${runProjectLinks.projectId} end)`;
|
|
range = resolveCostDateRange(range);
|
|
const conditions: ReturnType<typeof eq>[] = [eq(costEvents.companyId, companyId)];
|
|
if (range?.from) conditions.push(gte(costEvents.occurredAt, range.from));
|
|
if (range?.to) conditions.push(lte(costEvents.occurredAt, range.to));
|
|
|
|
const costCentsExpr = sumAsNumber(costEvents.costCents);
|
|
|
|
return db
|
|
.select({
|
|
projectId: projects.id,
|
|
projectName: projects.name,
|
|
eventCount: sql<number>`count(*)::int`,
|
|
estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
|
|
costCents: costCentsExpr,
|
|
costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
|
inputTokens: sumAsNumber(ordinaryInputTokens),
|
|
cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
|
|
outputTokens: sumAsNumber(costEvents.outputTokens),
|
|
})
|
|
.from(costEvents)
|
|
.leftJoin(runProjectLinks, eq(costEvents.heartbeatRunId, runProjectLinks.runId))
|
|
.leftJoin(projects, and(eq(projects.companyId, companyId), sql`${projects.id} = ${effectiveProjectId}`))
|
|
.where(and(...conditions))
|
|
.groupBy(projects.id, projects.name)
|
|
.orderBy(desc(costCentsExpr));
|
|
},
|
|
};
|
|
}
|