fix: localize downstream key trend timestamps (#432)
* fix: localize downstream key trend timestamps * fix: keep downstream trend metadata on empty responses * fix: cursor paginate downstream key trend reads * fix: normalize empty downstream trend timezones * fix: localize windowed downstream trend buckets
This commit is contained in:
Cita
@@ -41,11 +41,11 @@ describe('downstream api keys routes', () => {
|
||||
});
|
||||
|
||||
it('builds postgres trend buckets by casting text timestamps before date_trunc', async () => {
|
||||
const routesModule = await import('./downstreamApiKeys.js') as Record<string, any>;
|
||||
const trendServiceModule = await import('../../services/downstreamApiKeyTrendService.js') as Record<string, any>;
|
||||
|
||||
expect(typeof routesModule.buildBucketTsExpressionForDialect).toBe('function');
|
||||
expect(typeof trendServiceModule.buildBucketTsExpressionForDialect).toBe('function');
|
||||
|
||||
const expression = routesModule.buildBucketTsExpressionForDialect('postgres', schema.proxyLogs.createdAt, 3600);
|
||||
const expression = trendServiceModule.buildBucketTsExpressionForDialect('postgres', schema.proxyLogs.createdAt, 3600);
|
||||
const rendered = new PgDialect().sqlToQuery(expression).sql;
|
||||
|
||||
expect(rendered).toContain('date_trunc');
|
||||
@@ -787,6 +787,71 @@ describe('downstream api keys routes', () => {
|
||||
expect(trendBody.buckets.some((bucket: any) => bucket.totalTokens === 900)).toBe(true);
|
||||
});
|
||||
|
||||
it('groups all-range trend buckets by local day boundaries', async () => {
|
||||
const inserted = await db.insert(schema.downstreamApiKeys).values({
|
||||
name: 'local-day-key',
|
||||
key: 'sk-local-day-key-001',
|
||||
enabled: true,
|
||||
description: 'local day trend',
|
||||
usedCost: 0,
|
||||
usedRequests: 0,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.proxyLogs).values([
|
||||
{
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'success',
|
||||
totalTokens: 100,
|
||||
estimatedCost: 0.01,
|
||||
createdAt: '2026-04-05T23:30:00.000Z',
|
||||
},
|
||||
{
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'failed',
|
||||
totalTokens: 200,
|
||||
estimatedCost: 0.02,
|
||||
createdAt: '2026-04-06T01:00:00.000Z',
|
||||
},
|
||||
{
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'success',
|
||||
totalTokens: 300,
|
||||
estimatedCost: 0.03,
|
||||
createdAt: '2026-04-06T16:30:00.000Z',
|
||||
},
|
||||
]).run();
|
||||
|
||||
const trendRes = await app.inject({
|
||||
method: 'GET',
|
||||
url: `/api/downstream-keys/${inserted.id}/trend?range=all&timeZone=${encodeURIComponent('Asia/Shanghai')}`,
|
||||
});
|
||||
|
||||
expect(trendRes.statusCode).toBe(200);
|
||||
expect(trendRes.json()).toMatchObject({
|
||||
success: true,
|
||||
range: 'all',
|
||||
timeZone: 'Asia/Shanghai',
|
||||
buckets: [
|
||||
{
|
||||
startUtc: '2026-04-05T16:00:00.000Z',
|
||||
totalRequests: 2,
|
||||
successRequests: 1,
|
||||
failedRequests: 1,
|
||||
totalTokens: 300,
|
||||
totalCost: 0.03,
|
||||
},
|
||||
{
|
||||
startUtc: '2026-04-06T16:00:00.000Z',
|
||||
totalRequests: 1,
|
||||
successRequests: 1,
|
||||
failedRequests: 0,
|
||||
totalTokens: 300,
|
||||
totalCost: 0.03,
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects unknown route ids and site ids in downstream policy payloads', async () => {
|
||||
const createRes = await app.inject({
|
||||
method: 'POST',
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { and, eq, inArray, sql, type SQL, type SQLWrapper } from 'drizzle-orm';
|
||||
import { and, eq, inArray, sql, type SQL } from 'drizzle-orm';
|
||||
import { db, hasProxyLogDownstreamApiKeyIdColumn, runtimeDbDialect, schema } from '../../db/index.js';
|
||||
import { insertAndGetById } from '../../db/insertHelpers.js';
|
||||
import {
|
||||
@@ -9,8 +9,15 @@ import {
|
||||
toDownstreamApiKeyPolicyView,
|
||||
toPersistenceJson,
|
||||
} from '../../services/downstreamApiKeyService.js';
|
||||
import type { DownstreamExcludedCredentialRef } from '../../services/downstreamPolicyTypes.js';
|
||||
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
|
||||
import type { DownstreamExcludedCredentialRef } from '../../services/downstreamPolicyTypes.js';
|
||||
import {
|
||||
readDownstreamApiKeyTrendBuckets,
|
||||
resolveDownstreamTrendBucketSeconds,
|
||||
resolveDownstreamTrendRangeSinceUtc,
|
||||
resolveDownstreamTrendTimeZone,
|
||||
type DownstreamKeyTrendRange,
|
||||
} from '../../services/downstreamApiKeyTrendService.js';
|
||||
import {
|
||||
parseDownstreamApiKeyBatchPayload,
|
||||
parseDownstreamApiKeyPayload,
|
||||
@@ -78,7 +85,7 @@ function normalizeBatchIds(raw: unknown): number[] {
|
||||
return ids;
|
||||
}
|
||||
|
||||
type DownstreamKeyRange = '24h' | '7d' | 'all';
|
||||
type DownstreamKeyRange = DownstreamKeyTrendRange;
|
||||
type DownstreamKeyStatus = 'all' | 'enabled' | 'disabled';
|
||||
|
||||
function normalizeDownstreamKeyRange(raw: unknown): DownstreamKeyRange {
|
||||
@@ -124,37 +131,7 @@ function normalizeTagMatchMode(raw: unknown): 'any' | 'all' {
|
||||
}
|
||||
|
||||
function resolveRangeSinceUtc(range: DownstreamKeyRange): string | null {
|
||||
const nowTs = Date.now();
|
||||
if (range === '24h') return formatUtcSqlDateTime(new Date(nowTs - 24 * 60 * 60 * 1000));
|
||||
if (range === '7d') return formatUtcSqlDateTime(new Date(nowTs - 7 * 24 * 60 * 60 * 1000));
|
||||
return null;
|
||||
}
|
||||
|
||||
function resolveBucketSeconds(range: DownstreamKeyRange): number {
|
||||
return range === 'all' ? 86400 : 3600;
|
||||
}
|
||||
|
||||
export function buildBucketTsExpressionForDialect(
|
||||
dialect: 'sqlite' | 'mysql' | 'postgres',
|
||||
createdAtColumn: SQLWrapper,
|
||||
bucketSeconds: number,
|
||||
) {
|
||||
if (dialect === 'mysql') {
|
||||
return sql<number>`floor(unix_timestamp(${createdAtColumn}) / ${bucketSeconds}) * ${bucketSeconds}`;
|
||||
}
|
||||
if (dialect === 'postgres') {
|
||||
const createdAtTimestamp = sql`cast(${createdAtColumn} as timestamp)`;
|
||||
if (bucketSeconds === 86400) {
|
||||
return sql<number>`extract(epoch from date_trunc('day', ${createdAtTimestamp}))::bigint`;
|
||||
}
|
||||
return sql<number>`extract(epoch from date_trunc('hour', ${createdAtTimestamp}))::bigint`;
|
||||
}
|
||||
// sqlite
|
||||
return sql<number>`cast(cast(strftime('%s', ${createdAtColumn}) as integer) / ${bucketSeconds} as integer) * ${bucketSeconds}`;
|
||||
}
|
||||
|
||||
function resolveBucketTsExpression(bucketSeconds: number) {
|
||||
return buildBucketTsExpressionForDialect(runtimeDbDialect, schema.proxyLogs.createdAt, bucketSeconds);
|
||||
return resolveDownstreamTrendRangeSinceUtc(range);
|
||||
}
|
||||
|
||||
async function validatePolicyReferences(input: {
|
||||
@@ -447,7 +424,7 @@ export async function downstreamApiKeysRoutes(app: FastifyInstance) {
|
||||
return { success: true, item, usage: { last24h, last7d, all } };
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: string }; Querystring: { range?: string } }>('/api/downstream-keys/:id/trend', async (request, reply) => {
|
||||
app.get<{ Params: { id: string }; Querystring: { range?: string; timeZone?: string } }>('/api/downstream-keys/:id/trend', async (request, reply) => {
|
||||
const id = parseRouteId(request.params.id);
|
||||
if (!id) {
|
||||
return reply.code(400).send({ success: false, message: 'id 无效' });
|
||||
@@ -461,49 +438,29 @@ export async function downstreamApiKeysRoutes(app: FastifyInstance) {
|
||||
|
||||
const columnReady = await hasProxyLogDownstreamApiKeyIdColumn();
|
||||
if (!columnReady) {
|
||||
return { success: true, range, item: { id: item.id, name: item.name }, buckets: [] };
|
||||
return {
|
||||
success: true,
|
||||
range,
|
||||
item: { id: item.id, name: item.name },
|
||||
bucketSeconds: resolveDownstreamTrendBucketSeconds(range),
|
||||
timeZone: resolveDownstreamTrendTimeZone(request.query?.timeZone),
|
||||
buckets: [],
|
||||
};
|
||||
}
|
||||
|
||||
const bucketSeconds = resolveBucketSeconds(range);
|
||||
const bucketTs = resolveBucketTsExpression(bucketSeconds);
|
||||
const sinceUtc = resolveRangeSinceUtc(range);
|
||||
|
||||
const rows = await db.select({
|
||||
bucketTs,
|
||||
totalRequests: sql<number>`count(*)`,
|
||||
successRequests: sql<number>`coalesce(sum(case when ${schema.proxyLogs.status} = 'success' then 1 else 0 end), 0)`,
|
||||
failedRequests: sql<number>`coalesce(sum(case when ${schema.proxyLogs.status} = 'success' then 0 else 1 end), 0)`,
|
||||
totalTokens: sql<number>`coalesce(sum(coalesce(${schema.proxyLogs.totalTokens}, 0)), 0)`,
|
||||
totalCost: sql<number>`coalesce(sum(coalesce(${schema.proxyLogs.estimatedCost}, 0)), 0)`,
|
||||
})
|
||||
.from(schema.proxyLogs)
|
||||
.where(and(
|
||||
eq(schema.proxyLogs.downstreamApiKeyId, id),
|
||||
...(sinceUtc ? [sql`${schema.proxyLogs.createdAt} >= ${sinceUtc}`] : []),
|
||||
))
|
||||
.groupBy(bucketTs)
|
||||
.orderBy(bucketTs)
|
||||
.all();
|
||||
const trend = await readDownstreamApiKeyTrendBuckets({
|
||||
downstreamApiKeyId: id,
|
||||
range,
|
||||
timeZone: request.query?.timeZone,
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
range,
|
||||
item: { id: item.id, name: item.name },
|
||||
bucketSeconds,
|
||||
buckets: rows.map((row: any) => {
|
||||
const tsSeconds = Number(row.bucketTs || 0);
|
||||
const totalRequests = Number(row.totalRequests || 0);
|
||||
const successRequests = Number(row.successRequests || 0);
|
||||
return {
|
||||
startUtc: tsSeconds > 0 ? new Date(tsSeconds * 1000).toISOString() : null,
|
||||
totalRequests,
|
||||
successRequests,
|
||||
failedRequests: Number(row.failedRequests || 0),
|
||||
successRate: totalRequests > 0 ? Math.round((successRequests / totalRequests) * 1000) / 10 : null,
|
||||
totalTokens: Number(row.totalTokens || 0),
|
||||
totalCost: Math.round(Number(row.totalCost || 0) * 1_000_000) / 1_000_000,
|
||||
};
|
||||
}),
|
||||
bucketSeconds: trend.bucketSeconds,
|
||||
timeZone: trend.timeZone,
|
||||
buckets: trend.buckets,
|
||||
};
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
|
||||
import { mkdtempSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
|
||||
type DbModule = typeof import('../db/index.js');
|
||||
type TrendServiceModule = typeof import('./downstreamApiKeyTrendService.js');
|
||||
|
||||
const INSERT_BATCH_SIZE = 100;
|
||||
|
||||
describe('downstreamApiKeyTrendService', () => {
|
||||
let db: DbModule['db'];
|
||||
let schema: DbModule['schema'];
|
||||
let closeDbConnections: DbModule['closeDbConnections'];
|
||||
let trendService: TrendServiceModule;
|
||||
let dataDir = '';
|
||||
|
||||
beforeAll(async () => {
|
||||
dataDir = mkdtempSync(join(tmpdir(), 'metapi-downstream-trend-'));
|
||||
process.env.DATA_DIR = dataDir;
|
||||
|
||||
await import('../db/migrate.js');
|
||||
const dbModule = await import('../db/index.js');
|
||||
const trendServiceModule = await import('./downstreamApiKeyTrendService.js');
|
||||
|
||||
db = dbModule.db;
|
||||
schema = dbModule.schema;
|
||||
closeDbConnections = dbModule.closeDbConnections;
|
||||
trendService = trendServiceModule;
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await db.delete(schema.proxyLogs).run();
|
||||
await db.delete(schema.downstreamApiKeys).run();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await closeDbConnections();
|
||||
delete process.env.DATA_DIR;
|
||||
});
|
||||
|
||||
it('reads all-range buckets across cursor pages when many rows share the same createdAt', async () => {
|
||||
const inserted = await db.insert(schema.downstreamApiKeys).values({
|
||||
name: 'cursor-key',
|
||||
key: 'sk-cursor-key-001',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
const sharedCreatedAt = '2026-04-05T00:15:00.000Z';
|
||||
const sharedRows = Array.from({ length: 5_001 }, () => ({
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'success',
|
||||
totalTokens: 1,
|
||||
estimatedCost: 0.001,
|
||||
createdAt: sharedCreatedAt,
|
||||
}));
|
||||
|
||||
for (let index = 0; index < sharedRows.length; index += INSERT_BATCH_SIZE) {
|
||||
await db.insert(schema.proxyLogs).values(sharedRows.slice(index, index + INSERT_BATCH_SIZE)).run();
|
||||
}
|
||||
|
||||
await db.insert(schema.proxyLogs).values({
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'failed',
|
||||
totalTokens: 2,
|
||||
estimatedCost: 0.002,
|
||||
createdAt: '2026-04-06T00:30:00.000Z',
|
||||
}).run();
|
||||
|
||||
const trend = await trendService.readDownstreamApiKeyTrendBuckets({
|
||||
downstreamApiKeyId: inserted.id,
|
||||
range: 'all',
|
||||
timeZone: 'UTC',
|
||||
});
|
||||
|
||||
expect(trend.bucketSeconds).toBe(86400);
|
||||
expect(trend.timeZone).toBe('UTC');
|
||||
expect(trend.buckets).toHaveLength(2);
|
||||
expect(trend.buckets[0]).toMatchObject({
|
||||
startUtc: '2026-04-05T00:00:00.000Z',
|
||||
totalRequests: 5_001,
|
||||
successRequests: 5_001,
|
||||
failedRequests: 0,
|
||||
totalTokens: 5_001,
|
||||
});
|
||||
expect(trend.buckets[0]?.totalCost).toBeCloseTo(5.001, 6);
|
||||
expect(trend.buckets[1]).toMatchObject({
|
||||
startUtc: '2026-04-06T00:00:00.000Z',
|
||||
totalRequests: 1,
|
||||
successRequests: 0,
|
||||
failedRequests: 1,
|
||||
totalTokens: 2,
|
||||
});
|
||||
expect(trend.buckets[1]?.totalCost).toBeCloseTo(0.002, 6);
|
||||
});
|
||||
|
||||
it('normalizes trend time zones consistently for explicit and invalid values', () => {
|
||||
const fallback = trendService.resolveDownstreamTrendTimeZone();
|
||||
|
||||
expect(trendService.resolveDownstreamTrendTimeZone('UTC')).toBe('UTC');
|
||||
expect(trendService.resolveDownstreamTrendTimeZone('Invalid/Zone')).toBe(fallback);
|
||||
expect(trendService.resolveDownstreamTrendTimeZone(' ')).toBe(fallback);
|
||||
});
|
||||
|
||||
it('uses local hour buckets for windowed ranges in half-hour offset time zones', async () => {
|
||||
const inserted = await db.insert(schema.downstreamApiKeys).values({
|
||||
name: 'windowed-local-hour-key',
|
||||
key: 'sk-windowed-local-hour-key-001',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
const baseHour = new Date(Date.now() - 2 * 60 * 60 * 1000);
|
||||
baseHour.setUTCMinutes(0, 0, 0);
|
||||
const firstCreatedAt = new Date(baseHour);
|
||||
firstCreatedAt.setUTCMinutes(10, 0, 0);
|
||||
const secondCreatedAt = new Date(baseHour);
|
||||
secondCreatedAt.setUTCMinutes(40, 0, 0);
|
||||
|
||||
await db.insert(schema.proxyLogs).values([
|
||||
{
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'success',
|
||||
totalTokens: 10,
|
||||
estimatedCost: 0.01,
|
||||
createdAt: firstCreatedAt.toISOString(),
|
||||
},
|
||||
{
|
||||
downstreamApiKeyId: inserted.id,
|
||||
status: 'failed',
|
||||
totalTokens: 20,
|
||||
estimatedCost: 0.02,
|
||||
createdAt: secondCreatedAt.toISOString(),
|
||||
},
|
||||
]).run();
|
||||
|
||||
const trend = await trendService.readDownstreamApiKeyTrendBuckets({
|
||||
downstreamApiKeyId: inserted.id,
|
||||
range: '24h',
|
||||
timeZone: 'Asia/Kolkata',
|
||||
});
|
||||
|
||||
expect(trend.bucketSeconds).toBe(3600);
|
||||
expect(trend.timeZone).toBe('Asia/Kolkata');
|
||||
expect(trend.buckets).toMatchObject([
|
||||
{
|
||||
startUtc: expectedFixedOffsetHourBucketStartUtc(firstCreatedAt.toISOString(), 330),
|
||||
totalRequests: 1,
|
||||
successRequests: 1,
|
||||
failedRequests: 0,
|
||||
totalTokens: 10,
|
||||
totalCost: 0.01,
|
||||
},
|
||||
{
|
||||
startUtc: expectedFixedOffsetHourBucketStartUtc(secondCreatedAt.toISOString(), 330),
|
||||
totalRequests: 1,
|
||||
successRequests: 0,
|
||||
failedRequests: 1,
|
||||
totalTokens: 20,
|
||||
totalCost: 0.02,
|
||||
},
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
function expectedFixedOffsetHourBucketStartUtc(raw: string, offsetMinutes: number): string {
|
||||
const parsed = new Date(raw);
|
||||
const localMs = parsed.getTime() + offsetMinutes * 60_000;
|
||||
const localBucketStart = new Date(localMs);
|
||||
localBucketStart.setUTCMinutes(0, 0, 0);
|
||||
return new Date(localBucketStart.getTime() - offsetMinutes * 60_000).toISOString();
|
||||
}
|
||||
@@ -0,0 +1,334 @@
|
||||
import { and, asc, eq, sql, type SQL, type SQLWrapper } from 'drizzle-orm';
|
||||
import { db, runtimeDbDialect, schema } from '../db/index.js';
|
||||
import {
|
||||
formatUtcSqlDateTime,
|
||||
getResolvedTimeZone,
|
||||
parseStoredUtcDateTime,
|
||||
type StoredUtcDateTimeInput,
|
||||
} from './localTimeService.js';
|
||||
|
||||
export type DownstreamKeyTrendRange = '24h' | '7d' | 'all';
|
||||
|
||||
export type DownstreamKeyTrendBucket = {
|
||||
startUtc: string | null;
|
||||
totalRequests: number;
|
||||
successRequests: number;
|
||||
failedRequests: number;
|
||||
successRate: number | null;
|
||||
totalTokens: number;
|
||||
totalCost: number;
|
||||
};
|
||||
|
||||
type DownstreamTrendLogRow = {
|
||||
id: number;
|
||||
createdAt: StoredUtcDateTimeInput;
|
||||
status: string | null;
|
||||
totalTokens: number | null;
|
||||
totalCost: number | null;
|
||||
};
|
||||
|
||||
type DownstreamTrendBucketAccumulator = {
|
||||
startUtc: string;
|
||||
totalRequests: number;
|
||||
successRequests: number;
|
||||
failedRequests: number;
|
||||
totalTokens: number;
|
||||
totalCost: number;
|
||||
};
|
||||
|
||||
type TimeZoneDateTimeParts = {
|
||||
year: number;
|
||||
month: number;
|
||||
day: number;
|
||||
hour: number;
|
||||
minute: number;
|
||||
second: number;
|
||||
};
|
||||
|
||||
type DownstreamTrendCursor = {
|
||||
createdAt: StoredUtcDateTimeInput;
|
||||
id: number;
|
||||
};
|
||||
|
||||
const ALL_RANGE_CHUNK_SIZE = 5_000;
|
||||
|
||||
export function resolveDownstreamTrendRangeSinceUtc(range: DownstreamKeyTrendRange): string | null {
|
||||
const nowTs = Date.now();
|
||||
if (range === '24h') return formatUtcSqlDateTime(new Date(nowTs - 24 * 60 * 60 * 1000));
|
||||
if (range === '7d') return formatUtcSqlDateTime(new Date(nowTs - 7 * 24 * 60 * 60 * 1000));
|
||||
return null;
|
||||
}
|
||||
|
||||
export function resolveDownstreamTrendBucketSeconds(range: DownstreamKeyTrendRange): number {
|
||||
return range === 'all' ? 86400 : 3600;
|
||||
}
|
||||
|
||||
export function buildBucketTsExpressionForDialect(
|
||||
dialect: 'sqlite' | 'mysql' | 'postgres',
|
||||
createdAtColumn: SQLWrapper,
|
||||
bucketSeconds: number,
|
||||
) {
|
||||
if (dialect === 'mysql') {
|
||||
return sql<number>`floor(unix_timestamp(${createdAtColumn}) / ${bucketSeconds}) * ${bucketSeconds}`;
|
||||
}
|
||||
if (dialect === 'postgres') {
|
||||
const createdAtTimestamp = sql`cast(${createdAtColumn} as timestamp)`;
|
||||
if (bucketSeconds === 86400) {
|
||||
return sql<number>`extract(epoch from date_trunc('day', ${createdAtTimestamp}))::bigint`;
|
||||
}
|
||||
return sql<number>`extract(epoch from date_trunc('hour', ${createdAtTimestamp}))::bigint`;
|
||||
}
|
||||
return sql<number>`cast(cast(strftime('%s', ${createdAtColumn}) as integer) / ${bucketSeconds} as integer) * ${bucketSeconds}`;
|
||||
}
|
||||
|
||||
function resolveBucketTsExpression(bucketSeconds: number) {
|
||||
return buildBucketTsExpressionForDialect(runtimeDbDialect, schema.proxyLogs.createdAt, bucketSeconds);
|
||||
}
|
||||
|
||||
export function resolveDownstreamTrendTimeZone(raw?: string | null): string {
|
||||
const text = String(raw || '').trim();
|
||||
if (text) {
|
||||
try {
|
||||
new Intl.DateTimeFormat('en-CA', { timeZone: text }).format(new Date());
|
||||
return text;
|
||||
} catch {
|
||||
// fall through to server default
|
||||
}
|
||||
}
|
||||
return getResolvedTimeZone();
|
||||
}
|
||||
|
||||
function formatDateTimePartsInTimeZone(value: Date, timeZone: string): TimeZoneDateTimeParts | null {
|
||||
if (Number.isNaN(value.getTime())) return null;
|
||||
const formatter = new Intl.DateTimeFormat('en-CA', {
|
||||
timeZone,
|
||||
year: 'numeric',
|
||||
month: '2-digit',
|
||||
day: '2-digit',
|
||||
hour: '2-digit',
|
||||
minute: '2-digit',
|
||||
second: '2-digit',
|
||||
hour12: false,
|
||||
});
|
||||
const parts = formatter.formatToParts(value);
|
||||
const lookup = new Map(parts.map((part) => [part.type, part.value]));
|
||||
const year = Number(lookup.get('year'));
|
||||
const month = Number(lookup.get('month'));
|
||||
const day = Number(lookup.get('day'));
|
||||
const hour = Number(lookup.get('hour'));
|
||||
const minute = Number(lookup.get('minute'));
|
||||
const second = Number(lookup.get('second'));
|
||||
if (![year, month, day, hour, minute, second].every(Number.isFinite)) return null;
|
||||
return { year, month, day, hour, minute, second };
|
||||
}
|
||||
|
||||
function toUtcComparableValue(parts: TimeZoneDateTimeParts): number {
|
||||
return Date.UTC(parts.year, parts.month - 1, parts.day, parts.hour, parts.minute, parts.second, 0);
|
||||
}
|
||||
|
||||
function sameDateTimeParts(left: TimeZoneDateTimeParts, right: TimeZoneDateTimeParts): boolean {
|
||||
return left.year === right.year
|
||||
&& left.month === right.month
|
||||
&& left.day === right.day
|
||||
&& left.hour === right.hour
|
||||
&& left.minute === right.minute
|
||||
&& left.second === right.second;
|
||||
}
|
||||
|
||||
function resolveTargetBucketParts(parts: TimeZoneDateTimeParts, bucketSeconds: number): TimeZoneDateTimeParts {
|
||||
if (bucketSeconds >= 86400) {
|
||||
return { ...parts, hour: 0, minute: 0, second: 0 };
|
||||
}
|
||||
return { ...parts, minute: 0, second: 0 };
|
||||
}
|
||||
|
||||
function convertZonedPartsToUtcDate(parts: TimeZoneDateTimeParts, timeZone: string): Date | null {
|
||||
let candidateMs = toUtcComparableValue(parts);
|
||||
// Iterate toward the UTC instant that renders back to the requested local wall time.
|
||||
// Rare DST gaps/ambiguities may never converge; callers skip null buckets safely.
|
||||
for (let attempt = 0; attempt < 4; attempt += 1) {
|
||||
const candidate = new Date(candidateMs);
|
||||
const rendered = formatDateTimePartsInTimeZone(candidate, timeZone);
|
||||
if (!rendered) return null;
|
||||
if (sameDateTimeParts(rendered, parts)) return candidate;
|
||||
candidateMs += toUtcComparableValue(parts) - toUtcComparableValue(rendered);
|
||||
}
|
||||
const finalCandidate = new Date(candidateMs);
|
||||
const finalRendered = formatDateTimePartsInTimeZone(finalCandidate, timeZone);
|
||||
return finalRendered && sameDateTimeParts(finalRendered, parts) ? finalCandidate : null;
|
||||
}
|
||||
|
||||
function resolveLocalBucketStartUtc(
|
||||
raw: StoredUtcDateTimeInput,
|
||||
bucketSeconds: number,
|
||||
timeZone: string,
|
||||
): string | null {
|
||||
const parsed = parseStoredUtcDateTime(raw);
|
||||
if (!parsed) return null;
|
||||
const localParts = formatDateTimePartsInTimeZone(parsed, timeZone);
|
||||
if (!localParts) return null;
|
||||
const targetParts = resolveTargetBucketParts(localParts, bucketSeconds);
|
||||
const bucketStart = convertZonedPartsToUtcDate(targetParts, timeZone);
|
||||
return bucketStart ? bucketStart.toISOString() : null;
|
||||
}
|
||||
|
||||
function accumulateTrendRows(
|
||||
accumulator: Map<string, DownstreamTrendBucketAccumulator>,
|
||||
rows: DownstreamTrendLogRow[],
|
||||
bucketSeconds: number,
|
||||
timeZone: string,
|
||||
) {
|
||||
for (const row of rows) {
|
||||
const startUtc = resolveLocalBucketStartUtc(row.createdAt, bucketSeconds, timeZone);
|
||||
if (!startUtc) continue;
|
||||
const bucket = accumulator.get(startUtc) || {
|
||||
startUtc,
|
||||
totalRequests: 0,
|
||||
successRequests: 0,
|
||||
failedRequests: 0,
|
||||
totalTokens: 0,
|
||||
totalCost: 0,
|
||||
};
|
||||
const isSuccess = (row.status || '').trim().toLowerCase() === 'success';
|
||||
bucket.totalRequests += 1;
|
||||
bucket.successRequests += isSuccess ? 1 : 0;
|
||||
bucket.failedRequests += isSuccess ? 0 : 1;
|
||||
bucket.totalTokens += Number(row.totalTokens || 0);
|
||||
bucket.totalCost += Number(row.totalCost || 0);
|
||||
accumulator.set(startUtc, bucket);
|
||||
}
|
||||
}
|
||||
|
||||
function finalizeTrendBuckets(accumulator: Map<string, DownstreamTrendBucketAccumulator>): DownstreamKeyTrendBucket[] {
|
||||
return Array.from(accumulator.values())
|
||||
.sort((left, right) => left.startUtc.localeCompare(right.startUtc))
|
||||
.map((bucket) => ({
|
||||
startUtc: bucket.startUtc,
|
||||
totalRequests: bucket.totalRequests,
|
||||
successRequests: bucket.successRequests,
|
||||
failedRequests: bucket.failedRequests,
|
||||
successRate: bucket.totalRequests > 0 ? Math.round((bucket.successRequests / bucket.totalRequests) * 1000) / 10 : null,
|
||||
totalTokens: bucket.totalTokens,
|
||||
totalCost: Math.round(bucket.totalCost * 1_000_000) / 1_000_000,
|
||||
}));
|
||||
}
|
||||
|
||||
function buildTrendCursorClause(cursor: DownstreamTrendCursor): SQL {
|
||||
return sql`(
|
||||
${schema.proxyLogs.createdAt} > ${cursor.createdAt}
|
||||
or (${schema.proxyLogs.createdAt} = ${cursor.createdAt} and ${schema.proxyLogs.id} > ${cursor.id})
|
||||
)`;
|
||||
}
|
||||
|
||||
async function readAllRangeTrendBuckets(
|
||||
downstreamApiKeyId: number,
|
||||
bucketSeconds: number,
|
||||
timeZone: string,
|
||||
sinceUtc: string | null,
|
||||
): Promise<DownstreamKeyTrendBucket[]> {
|
||||
const accumulator = new Map<string, DownstreamTrendBucketAccumulator>();
|
||||
let cursor: DownstreamTrendCursor | null = null;
|
||||
|
||||
for (;;) {
|
||||
const whereClauses: SQL[] = [eq(schema.proxyLogs.downstreamApiKeyId, downstreamApiKeyId)];
|
||||
if (sinceUtc) {
|
||||
whereClauses.push(sql`${schema.proxyLogs.createdAt} >= ${sinceUtc}`);
|
||||
}
|
||||
if (cursor) {
|
||||
whereClauses.push(buildTrendCursorClause(cursor));
|
||||
}
|
||||
|
||||
const rows = await db.select({
|
||||
id: schema.proxyLogs.id,
|
||||
createdAt: schema.proxyLogs.createdAt,
|
||||
status: schema.proxyLogs.status,
|
||||
totalTokens: schema.proxyLogs.totalTokens,
|
||||
totalCost: schema.proxyLogs.estimatedCost,
|
||||
})
|
||||
.from(schema.proxyLogs)
|
||||
.where(and(...whereClauses))
|
||||
.orderBy(asc(schema.proxyLogs.createdAt), asc(schema.proxyLogs.id))
|
||||
.limit(ALL_RANGE_CHUNK_SIZE)
|
||||
.all();
|
||||
|
||||
if (rows.length <= 0) break;
|
||||
accumulateTrendRows(accumulator, rows, bucketSeconds, timeZone);
|
||||
const lastRow = rows.at(-1);
|
||||
if (rows.length < ALL_RANGE_CHUNK_SIZE || !lastRow) break;
|
||||
cursor = {
|
||||
createdAt: lastRow.createdAt,
|
||||
id: lastRow.id,
|
||||
};
|
||||
}
|
||||
|
||||
return finalizeTrendBuckets(accumulator);
|
||||
}
|
||||
|
||||
async function readWindowedTrendBuckets(
|
||||
downstreamApiKeyId: number,
|
||||
bucketSeconds: number,
|
||||
sinceUtc: string | null,
|
||||
timeZone: string,
|
||||
): Promise<DownstreamKeyTrendBucket[]> {
|
||||
if (timeZone.toUpperCase() !== 'UTC') {
|
||||
return readAllRangeTrendBuckets(downstreamApiKeyId, bucketSeconds, timeZone, sinceUtc);
|
||||
}
|
||||
|
||||
const bucketTs = resolveBucketTsExpression(bucketSeconds);
|
||||
const whereClauses: SQL[] = [eq(schema.proxyLogs.downstreamApiKeyId, downstreamApiKeyId)];
|
||||
if (sinceUtc) {
|
||||
whereClauses.push(sql`${schema.proxyLogs.createdAt} >= ${sinceUtc}`);
|
||||
}
|
||||
|
||||
const rows = await db.select({
|
||||
bucketTs,
|
||||
totalRequests: sql<number>`count(*)`,
|
||||
successRequests: sql<number>`coalesce(sum(case when ${schema.proxyLogs.status} = 'success' then 1 else 0 end), 0)`,
|
||||
failedRequests: sql<number>`coalesce(sum(case when ${schema.proxyLogs.status} = 'success' then 0 else 1 end), 0)`,
|
||||
totalTokens: sql<number>`coalesce(sum(coalesce(${schema.proxyLogs.totalTokens}, 0)), 0)`,
|
||||
totalCost: sql<number>`coalesce(sum(coalesce(${schema.proxyLogs.estimatedCost}, 0)), 0)`,
|
||||
})
|
||||
.from(schema.proxyLogs)
|
||||
.where(and(...whereClauses))
|
||||
.groupBy(bucketTs)
|
||||
.orderBy(bucketTs)
|
||||
.all();
|
||||
|
||||
return rows.map((row: any) => {
|
||||
const tsSeconds = Number(row.bucketTs || 0);
|
||||
const totalRequests = Number(row.totalRequests || 0);
|
||||
const successRequests = Number(row.successRequests || 0);
|
||||
return {
|
||||
startUtc: tsSeconds > 0 ? new Date(tsSeconds * 1000).toISOString() : null,
|
||||
totalRequests,
|
||||
successRequests,
|
||||
failedRequests: Number(row.failedRequests || 0),
|
||||
successRate: totalRequests > 0 ? Math.round((successRequests / totalRequests) * 1000) / 10 : null,
|
||||
totalTokens: Number(row.totalTokens || 0),
|
||||
totalCost: Math.round(Number(row.totalCost || 0) * 1_000_000) / 1_000_000,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export async function readDownstreamApiKeyTrendBuckets(input: {
|
||||
downstreamApiKeyId: number;
|
||||
range: DownstreamKeyTrendRange;
|
||||
timeZone?: string | null;
|
||||
}): Promise<{
|
||||
bucketSeconds: number;
|
||||
timeZone: string;
|
||||
buckets: DownstreamKeyTrendBucket[];
|
||||
}> {
|
||||
const bucketSeconds = resolveDownstreamTrendBucketSeconds(input.range);
|
||||
const sinceUtc = resolveDownstreamTrendRangeSinceUtc(input.range);
|
||||
const timeZone = resolveDownstreamTrendTimeZone(input.timeZone);
|
||||
const buckets = input.range === 'all'
|
||||
? await readAllRangeTrendBuckets(input.downstreamApiKeyId, bucketSeconds, timeZone, sinceUtc)
|
||||
: await readWindowedTrendBuckets(input.downstreamApiKeyId, bucketSeconds, sinceUtc, timeZone);
|
||||
|
||||
return {
|
||||
bucketSeconds,
|
||||
timeZone,
|
||||
buckets,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user