feat: add site headers editor and scheduled log cleanup

This commit is contained in:
bnvnvnv
2026-03-12 08:12:29 +00:00
parent 1da3608d42
commit b6da71b1d6
34 changed files with 1572 additions and 210 deletions
@@ -49,7 +49,12 @@ describe('backupService', () => {
name: 'roundtrip-site',
url: 'https://roundtrip.example.com',
platform: 'new-api',
externalCheckinUrl: 'https://checkin.roundtrip.example.com',
proxyUrl: 'http://127.0.0.1:8080',
useSystemProxy: true,
customHeaders: JSON.stringify({
'cf-access-client-id': 'roundtrip-client',
}),
status: 'active',
isPinned: true,
sortOrder: 9,
@@ -128,6 +133,9 @@ describe('backupService', () => {
const restoredChannel = await db.select().from(schema.routeChannels).where(eq(schema.routeChannels.routeId, route.id)).get();
expect(restoredSite?.proxyUrl).toBe('http://127.0.0.1:8080');
expect(restoredSite?.externalCheckinUrl).toBe('https://checkin.roundtrip.example.com');
expect(restoredSite?.useSystemProxy).toBe(true);
expect(restoredSite?.customHeaders).toBe('{"cf-access-client-id":"roundtrip-client"}');
expect(restoredSite?.isPinned).toBe(true);
expect(restoredSite?.sortOrder).toBe(9);
+13 -1
View File
@@ -164,6 +164,7 @@ function buildAccountsSectionFromRefBackup(data: RawBackupData): AccountsBackupS
platform,
proxyUrl: null,
useSystemProxy: false,
customHeaders: null,
status: 'active',
isPinned: false,
sortOrder: sites.length,
@@ -293,10 +294,18 @@ function isFiniteNumber(value: unknown): value is number {
}
function isSettingValueAcceptable(key: string, value: unknown): boolean {
if (key === 'checkin_cron' || key === 'balance_refresh_cron') {
if (key === 'checkin_cron' || key === 'balance_refresh_cron' || key === 'log_cleanup_cron') {
return typeof value === 'string' && cron.validate(value);
}
if (key === 'log_cleanup_usage_logs_enabled' || key === 'log_cleanup_program_logs_enabled') {
return typeof value === 'boolean';
}
if (key === 'log_cleanup_retention_days') {
return isFiniteNumber(value) && value >= 1;
}
if (key === 'proxy_token') {
return typeof value === 'string'
&& value.trim().length >= 6
@@ -465,8 +474,11 @@ async function importAccountsSection(section: AccountsBackupSection): Promise<vo
id: row.id,
name: row.name,
url: row.url,
externalCheckinUrl: row.externalCheckinUrl ?? null,
platform: row.platform,
proxyUrl: row.proxyUrl ?? null,
useSystemProxy: row.useSystemProxy ?? false,
customHeaders: row.customHeaders ?? null,
status: row.status || 'active',
isPinned: row.isPinned ?? false,
sortOrder: row.sortOrder ?? 0,
+3 -4
View File
@@ -14,7 +14,7 @@ import { decryptAccountPassword } from './accountCredentialService.js';
import { extractRuntimeHealth, setAccountRuntimeHealth } from './accountHealthService.js';
import { updateTodayIncomeSnapshot } from './todayIncomeRewardService.js';
import type { BalanceInfo } from './platforms/base.js';
import { resolveProxyUrlForSite, withExplicitProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js';
import { withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js';
function isSiteDisabled(status?: string | null): boolean {
return (status || 'active') === 'disabled';
@@ -237,7 +237,6 @@ async function fetchTodayIncomeFromLogs(params: {
accessToken: string;
platform?: string | null;
platformUserId?: number;
proxyUrl?: string | null;
}): Promise<number | null> {
const baseUrl = params.baseUrl.trim();
const accessToken = params.accessToken.trim();
@@ -272,7 +271,8 @@ async function fetchTodayIncomeFromLogs(params: {
});
try {
const response = await fetch(`${baseUrl}/api/log/self?${query.toString()}`, withExplicitProxyRequestInit(params.proxyUrl, {
const requestUrl = `${baseUrl}/api/log/self?${query.toString()}`;
const response = await fetch(requestUrl, await withSiteProxyRequestInit(requestUrl, {
method: 'GET',
headers,
}));
@@ -466,7 +466,6 @@ export async function refreshBalance(accountId: number) {
accessToken: activeAccessToken,
platform: site.platform,
platformUserId,
proxyUrl: resolveProxyUrlForSite(site),
});
if (typeof fallbackIncome === 'number' && Number.isFinite(fallbackIncome)) {
balanceInfo.todayIncome = fallbackIncome;
+92 -2
View File
@@ -7,19 +7,26 @@ import { checkinAll } from './checkinService.js';
import { refreshModelsAndRebuildRoutes } from './modelService.js';
import { sendNotification } from './notifyService.js';
import { buildDailySummaryNotification, collectDailySummaryMetrics } from './dailySummaryService.js';
import { cleanupConfiguredLogs, normalizeLogCleanupRetentionDays } from './logCleanupService.js';
let checkinTask: cron.ScheduledTask | null = null;
let balanceTask: cron.ScheduledTask | null = null;
let dailySummaryTask: cron.ScheduledTask | null = null;
let logCleanupTask: cron.ScheduledTask | null = null;
const DAILY_SUMMARY_DEFAULT_CRON = '58 23 * * *';
const LOG_CLEANUP_DEFAULT_CRON = '0 6 * * *';
async function resolveCronSetting(settingKey: string, fallback: string): Promise<string> {
async function resolveJsonSetting<T>(
settingKey: string,
isValid: (value: unknown) => value is T,
fallback: T,
): Promise<T> {
try {
const row = await db.select().from(schema.settings).where(eq(schema.settings.key, settingKey)).get();
if (row?.value) {
const parsed = JSON.parse(row.value);
if (typeof parsed === 'string' && cron.validate(parsed)) {
if (isValid(parsed)) {
return parsed;
}
}
@@ -27,6 +34,22 @@ async function resolveCronSetting(settingKey: string, fallback: string): Promise
return fallback;
}
async function resolveCronSetting(settingKey: string, fallback: string): Promise<string> {
return resolveJsonSetting(settingKey, (value): value is string => typeof value === 'string' && cron.validate(value), fallback);
}
async function resolveBooleanSetting(settingKey: string, fallback: boolean): Promise<boolean> {
return resolveJsonSetting(settingKey, (value): value is boolean => typeof value === 'boolean', fallback);
}
async function resolvePositiveIntegerSetting(settingKey: string, fallback: number): Promise<number> {
return resolveJsonSetting(
settingKey,
(value): value is number => typeof value === 'number' && Number.isFinite(value) && value >= 1,
fallback,
);
}
function createCheckinTask(cronExpr: string) {
return cron.schedule(cronExpr, async () => {
console.log(`[Scheduler] Running check-in at ${new Date().toISOString()}`);
@@ -72,20 +95,67 @@ function createDailySummaryTask(cronExpr: string) {
});
}
function createLogCleanupTask(cronExpr: string) {
return cron.schedule(cronExpr, async () => {
if (!config.logCleanupConfigured) {
console.log('[Scheduler] Log cleanup skipped: legacy fallback mode is active');
return;
}
console.log(`[Scheduler] Running log cleanup at ${new Date().toISOString()}`);
try {
const result = await cleanupConfiguredLogs();
if (!result.enabled) {
console.log('[Scheduler] Log cleanup skipped: no log target enabled');
return;
}
console.log(
`[Scheduler] Log cleanup complete: usage=${result.usageLogsDeleted}, program=${result.programLogsDeleted}, cutoff=${result.cutoffUtc}`,
);
} catch (err) {
console.error('[Scheduler] Log cleanup error:', err);
}
});
}
export async function startScheduler() {
const activeCheckinCron = await resolveCronSetting('checkin_cron', config.checkinCron);
const activeBalanceCron = await resolveCronSetting('balance_refresh_cron', config.balanceRefreshCron);
const activeDailySummaryCron = await resolveCronSetting('daily_summary_cron', DAILY_SUMMARY_DEFAULT_CRON);
const activeLogCleanupCron = await resolveCronSetting('log_cleanup_cron', config.logCleanupCron || LOG_CLEANUP_DEFAULT_CRON);
const activeLogCleanupUsageLogsEnabled = await resolveBooleanSetting(
'log_cleanup_usage_logs_enabled',
config.logCleanupUsageLogsEnabled,
);
const activeLogCleanupProgramLogsEnabled = await resolveBooleanSetting(
'log_cleanup_program_logs_enabled',
config.logCleanupProgramLogsEnabled,
);
const activeLogCleanupRetentionDays = await resolvePositiveIntegerSetting(
'log_cleanup_retention_days',
normalizeLogCleanupRetentionDays(config.logCleanupRetentionDays),
);
config.checkinCron = activeCheckinCron;
config.balanceRefreshCron = activeBalanceCron;
config.logCleanupCron = activeLogCleanupCron;
config.logCleanupUsageLogsEnabled = activeLogCleanupUsageLogsEnabled;
config.logCleanupProgramLogsEnabled = activeLogCleanupProgramLogsEnabled;
config.logCleanupRetentionDays = activeLogCleanupRetentionDays;
checkinTask?.stop();
balanceTask?.stop();
dailySummaryTask?.stop();
logCleanupTask?.stop();
checkinTask = createCheckinTask(activeCheckinCron);
balanceTask = createBalanceTask(activeBalanceCron);
dailySummaryTask = createDailySummaryTask(activeDailySummaryCron);
logCleanupTask = createLogCleanupTask(activeLogCleanupCron);
console.log(`[Scheduler] Check-in cron: ${activeCheckinCron}`);
console.log(`[Scheduler] Balance refresh cron: ${activeBalanceCron}`);
console.log(`[Scheduler] Daily summary cron: ${activeDailySummaryCron}`);
console.log(
`[Scheduler] Log cleanup cron: ${activeLogCleanupCron} (configured=${config.logCleanupConfigured}, usage=${activeLogCleanupUsageLogsEnabled}, program=${activeLogCleanupProgramLogsEnabled}, retentionDays=${activeLogCleanupRetentionDays})`,
);
}
export function updateCheckinCron(cronExpr: string) {
@@ -101,3 +171,23 @@ export function updateBalanceRefreshCron(cronExpr: string) {
balanceTask?.stop();
balanceTask = createBalanceTask(cronExpr);
}
export function updateLogCleanupSettings(input: {
cronExpr?: string;
usageLogsEnabled?: boolean;
programLogsEnabled?: boolean;
retentionDays?: number;
}) {
const cronExpr = input.cronExpr ?? config.logCleanupCron;
if (!cron.validate(cronExpr)) throw new Error(`Invalid cron: ${cronExpr}`);
const retentionDays = normalizeLogCleanupRetentionDays(input.retentionDays ?? config.logCleanupRetentionDays);
config.logCleanupCron = cronExpr;
if (input.usageLogsEnabled !== undefined) config.logCleanupUsageLogsEnabled = !!input.usageLogsEnabled;
if (input.programLogsEnabled !== undefined) config.logCleanupProgramLogsEnabled = !!input.programLogsEnabled;
config.logCleanupRetentionDays = retentionDays;
logCleanupTask?.stop();
logCleanupTask = createLogCleanupTask(cronExpr);
}
@@ -101,7 +101,7 @@ describe('databaseMigrationService', () => {
expect(normalized.ssl).toBe(false);
});
it.each(['postgres', 'mysql', 'sqlite'] as const)('creates or patches sites schema with use_system_proxy for %s', async (dialect) => {
it.each(['postgres', 'mysql', 'sqlite'] as const)('creates or patches sites schema with use_system_proxy and custom_headers for %s', async (dialect) => {
const executedSql: string[] = [];
await __databaseMigrationServiceTestUtils.ensureSchema({
dialect,
@@ -118,7 +118,7 @@ describe('databaseMigrationService', () => {
}
if (sqlText.includes('pragma_table_info') || sqlText.includes('information_schema.columns')) {
const columnName = String(params[1] ?? sqlText.match(/name = '([^']+)'/)?.[1] ?? '');
return columnName === 'use_system_proxy' ? 0 : 1;
return columnName === 'use_system_proxy' || columnName === 'custom_headers' ? 0 : 1;
}
return 0;
},
@@ -126,8 +126,10 @@ describe('databaseMigrationService', () => {
});
const useSystemProxySql = executedSql.find((sqlText) => sqlText.includes('use_system_proxy'));
const customHeadersSql = executedSql.find((sqlText) => sqlText.includes('custom_headers'));
expect(useSystemProxySql).toContain('use_system_proxy');
expect(customHeadersSql).toContain('custom_headers');
});
it.each(['postgres', 'mysql'] as const)('patches token_routes decision snapshot columns for %s', async (dialect) => {
@@ -159,7 +161,7 @@ describe('databaseMigrationService', () => {
).toBe(true);
});
it('includes useSystemProxy when building site migration statements', () => {
it('includes useSystemProxy and customHeaders when building site migration statements', () => {
const statements = __databaseMigrationServiceTestUtils.buildStatements({
version: 'test',
timestamp: Date.now(),
@@ -170,6 +172,7 @@ describe('databaseMigrationService', () => {
url: 'https://example.com',
platform: 'openai',
useSystemProxy: true,
customHeaders: '{"x-site-scope":"internal"}',
status: 'active',
}],
accounts: [],
@@ -190,8 +193,11 @@ describe('databaseMigrationService', () => {
const siteStatement = statements.find((statement) => statement.table === 'sites');
const useSystemProxyIndex = siteStatement?.columns.indexOf('use_system_proxy') ?? -1;
const customHeadersIndex = siteStatement?.columns.indexOf('custom_headers') ?? -1;
expect(useSystemProxyIndex).toBeGreaterThanOrEqual(0);
expect(siteStatement?.values[useSystemProxyIndex]).toBe(true);
expect(customHeadersIndex).toBeGreaterThanOrEqual(0);
expect(siteStatement?.values[customHeadersIndex]).toBe('{"x-site-scope":"internal"}');
});
});
@@ -259,7 +259,7 @@ function buildStatements(snapshot: BackupSnapshot): InsertStatement[] {
for (const row of snapshot.accounts.sites) {
statements.push({
table: 'sites',
columns: ['id', 'name', 'url', 'external_checkin_url', 'platform', 'proxy_url', 'use_system_proxy', 'status', 'is_pinned', 'sort_order', 'global_weight', 'api_key', 'created_at', 'updated_at'],
columns: ['id', 'name', 'url', 'external_checkin_url', 'platform', 'proxy_url', 'use_system_proxy', 'custom_headers', 'status', 'is_pinned', 'sort_order', 'global_weight', 'api_key', 'created_at', 'updated_at'],
values: [
asNumber(row.id, 0),
asNullableString(row.name),
@@ -268,6 +268,7 @@ function buildStatements(snapshot: BackupSnapshot): InsertStatement[] {
asNullableString(row.platform),
asNullableString(row.proxyUrl),
asBoolean(row.useSystemProxy, false),
asNullableString(row.customHeaders),
asNullableString(row.status) ?? 'active',
asBoolean(row.isPinned, false),
asNumber(row.sortOrder, 0),
+13 -1
View File
@@ -1,7 +1,8 @@
import { buildConfig, config } from '../config.js';
import { db, schema, switchRuntimeDatabase } from '../db/index.js';
import { updateBalanceRefreshCron, updateCheckinCron } from './checkinScheduler.js';
import { updateBalanceRefreshCron, updateCheckinCron, updateLogCleanupSettings } from './checkinScheduler.js';
import { ensureDefaultSitesSeeded } from './defaultSiteSeedService.js';
import { startProxyLogRetentionService } from './proxyLogRetentionService.js';
import { invalidateSiteProxyCache } from './siteProxy.js';
export const FACTORY_RESET_ADMIN_TOKEN = 'change-me-admin-token';
@@ -38,8 +39,19 @@ function resetRuntimeConfigToInitialState() {
config.dbType = 'sqlite';
config.dbUrl = '';
config.dbSsl = false;
config.logCleanupConfigured = false;
config.logCleanupUsageLogsEnabled = config.proxyLogRetentionDays > 0;
config.logCleanupProgramLogsEnabled = false;
config.logCleanupRetentionDays = Math.max(1, Math.trunc(config.proxyLogRetentionDays || config.logCleanupRetentionDays || 30));
updateCheckinCron(config.checkinCron);
updateBalanceRefreshCron(config.balanceRefreshCron);
updateLogCleanupSettings({
cronExpr: config.logCleanupCron,
usageLogsEnabled: config.logCleanupUsageLogsEnabled,
programLogsEnabled: config.logCleanupProgramLogsEnabled,
retentionDays: config.logCleanupRetentionDays,
});
startProxyLogRetentionService();
invalidateSiteProxyCache();
}
@@ -0,0 +1,156 @@
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 ConfigModule = typeof import('../config.js');
type CleanupModule = typeof import('./logCleanupService.js');
describe('logCleanupService', () => {
let db: DbModule['db'];
let schema: DbModule['schema'];
let config: ConfigModule['config'];
let cleanupConfiguredLogs: CleanupModule['cleanupConfiguredLogs'];
let dataDir = '';
let originalConfig: {
logCleanupUsageLogsEnabled: boolean;
logCleanupProgramLogsEnabled: boolean;
logCleanupRetentionDays: number;
};
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-log-cleanup-'));
process.env.DATA_DIR = dataDir;
await import('../db/migrate.js');
const dbModule = await import('../db/index.js');
const configModule = await import('../config.js');
const cleanupModule = await import('./logCleanupService.js');
db = dbModule.db;
schema = dbModule.schema;
config = configModule.config;
cleanupConfiguredLogs = cleanupModule.cleanupConfiguredLogs;
originalConfig = {
logCleanupUsageLogsEnabled: config.logCleanupUsageLogsEnabled,
logCleanupProgramLogsEnabled: config.logCleanupProgramLogsEnabled,
logCleanupRetentionDays: config.logCleanupRetentionDays,
};
});
beforeEach(async () => {
await db.delete(schema.events).run();
await db.delete(schema.proxyLogs).run();
await db.delete(schema.accounts).run();
await db.delete(schema.sites).run();
config.logCleanupUsageLogsEnabled = false;
config.logCleanupProgramLogsEnabled = false;
config.logCleanupRetentionDays = 30;
});
afterAll(() => {
config.logCleanupUsageLogsEnabled = originalConfig.logCleanupUsageLogsEnabled;
config.logCleanupProgramLogsEnabled = originalConfig.logCleanupProgramLogsEnabled;
config.logCleanupRetentionDays = originalConfig.logCleanupRetentionDays;
delete process.env.DATA_DIR;
});
async function seedAccount() {
const site = await db.insert(schema.sites).values({
name: 'cleanup-site',
url: 'https://cleanup.example.com',
platform: 'new-api',
status: 'active',
}).returning().get();
return await db.insert(schema.accounts).values({
siteId: site.id,
username: 'cleanup-user',
accessToken: 'cleanup-access-token',
apiToken: 'cleanup-api-token',
status: 'active',
}).returning().get();
}
it('cleans usage logs and program logs older than retention days', async () => {
const account = await seedAccount();
await db.insert(schema.proxyLogs).values([
{
accountId: account.id,
modelRequested: 'gpt-4.1-mini',
status: 'success',
createdAt: '2026-02-01 08:00:00',
},
{
accountId: account.id,
modelRequested: 'gpt-4.1-mini',
status: 'success',
createdAt: '2026-03-10 08:00:00',
},
]).run();
await db.insert(schema.events).values([
{
type: 'status',
title: 'old event',
message: 'cleanup me',
createdAt: '2026-02-01 08:00:00',
},
{
type: 'status',
title: 'new event',
message: 'keep me',
createdAt: '2026-03-10 08:00:00',
},
]).run();
const result = await cleanupConfiguredLogs({
usageLogsEnabled: true,
programLogsEnabled: true,
retentionDays: 7,
nowMs: Date.parse('2026-03-12T00:00:00Z'),
});
expect(result.enabled).toBe(true);
expect(result.usageLogsDeleted).toBe(1);
expect(result.programLogsDeleted).toBe(1);
expect(result.totalDeleted).toBe(2);
const remainingProxyLogs = await db.select().from(schema.proxyLogs).all();
const remainingEvents = await db.select().from(schema.events).all();
expect(remainingProxyLogs).toHaveLength(1);
expect(remainingProxyLogs[0]?.createdAt).toBe('2026-03-10 08:00:00');
expect(remainingEvents).toHaveLength(1);
expect(remainingEvents[0]?.title).toBe('new event');
});
it('skips cleanup when no target is enabled', async () => {
const account = await seedAccount();
await db.insert(schema.proxyLogs).values({
accountId: account.id,
modelRequested: 'gpt-4.1-mini',
status: 'success',
createdAt: '2026-01-01 08:00:00',
}).run();
await db.insert(schema.events).values({
type: 'status',
title: 'still here',
createdAt: '2026-01-01 08:00:00',
}).run();
const result = await cleanupConfiguredLogs({
usageLogsEnabled: false,
programLogsEnabled: false,
retentionDays: 7,
nowMs: Date.parse('2026-03-12T00:00:00Z'),
});
expect(result.enabled).toBe(false);
expect(result.totalDeleted).toBe(0);
expect(await db.select().from(schema.proxyLogs).all()).toHaveLength(1);
expect(await db.select().from(schema.events).all()).toHaveLength(1);
});
});
+138
View File
@@ -0,0 +1,138 @@
import { lt } from 'drizzle-orm';
import { config } from '../config.js';
import { db, schema } from '../db/index.js';
import { formatUtcSqlDateTime } from './localTimeService.js';
const DAY_MS = 24 * 60 * 60 * 1000;
export type LogCleanupOptions = {
usageLogsEnabled?: boolean;
programLogsEnabled?: boolean;
retentionDays?: number;
nowMs?: number;
};
export type LogCleanupResult = {
enabled: boolean;
usageLogsEnabled: boolean;
programLogsEnabled: boolean;
retentionDays: number;
cutoffUtc: string | null;
usageLogsDeleted: number;
programLogsDeleted: number;
totalDeleted: number;
};
export function normalizeLogCleanupRetentionDays(value: unknown, fallback = 30): number {
const parsed = Number(value);
if (Number.isFinite(parsed) && parsed >= 1) return Math.trunc(parsed);
const fallbackParsed = Number(fallback);
if (Number.isFinite(fallbackParsed) && fallbackParsed >= 1) return Math.trunc(fallbackParsed);
return 30;
}
export function getLogCleanupCutoffUtc(retentionDays: number, nowMs = Date.now()): string | null {
const normalizedDays = normalizeLogCleanupRetentionDays(retentionDays);
return formatUtcSqlDateTime(new Date(nowMs - normalizedDays * DAY_MS));
}
export async function cleanupUsageLogs(retentionDays: number, nowMs = Date.now()): Promise<{
retentionDays: number;
cutoffUtc: string | null;
deleted: number;
}> {
const normalizedDays = normalizeLogCleanupRetentionDays(retentionDays);
const cutoffUtc = getLogCleanupCutoffUtc(normalizedDays, nowMs);
if (!cutoffUtc) {
return {
retentionDays: normalizedDays,
cutoffUtc: null,
deleted: 0,
};
}
const deleted = (
await db.delete(schema.proxyLogs)
.where(lt(schema.proxyLogs.createdAt, cutoffUtc))
.run()
).changes;
return {
retentionDays: normalizedDays,
cutoffUtc,
deleted,
};
}
export async function cleanupProgramLogs(retentionDays: number, nowMs = Date.now()): Promise<{
retentionDays: number;
cutoffUtc: string | null;
deleted: number;
}> {
const normalizedDays = normalizeLogCleanupRetentionDays(retentionDays);
const cutoffUtc = getLogCleanupCutoffUtc(normalizedDays, nowMs);
if (!cutoffUtc) {
return {
retentionDays: normalizedDays,
cutoffUtc: null,
deleted: 0,
};
}
const deleted = (
await db.delete(schema.events)
.where(lt(schema.events.createdAt, cutoffUtc))
.run()
).changes;
return {
retentionDays: normalizedDays,
cutoffUtc,
deleted,
};
}
export async function cleanupConfiguredLogs(options: LogCleanupOptions = {}): Promise<LogCleanupResult> {
const usageLogsEnabled = options.usageLogsEnabled ?? config.logCleanupUsageLogsEnabled;
const programLogsEnabled = options.programLogsEnabled ?? config.logCleanupProgramLogsEnabled;
const retentionDays = normalizeLogCleanupRetentionDays(
options.retentionDays ?? config.logCleanupRetentionDays,
config.logCleanupRetentionDays,
);
const nowMs = options.nowMs ?? Date.now();
const enabled = usageLogsEnabled || programLogsEnabled;
const cutoffUtc = enabled ? getLogCleanupCutoffUtc(retentionDays, nowMs) : null;
if (!enabled || !cutoffUtc) {
return {
enabled: false,
usageLogsEnabled,
programLogsEnabled,
retentionDays,
cutoffUtc,
usageLogsDeleted: 0,
programLogsDeleted: 0,
totalDeleted: 0,
};
}
const usageResult = usageLogsEnabled
? await cleanupUsageLogs(retentionDays, nowMs)
: { deleted: 0 };
const programResult = programLogsEnabled
? await cleanupProgramLogs(retentionDays, nowMs)
: { deleted: 0 };
return {
enabled: true,
usageLogsEnabled,
programLogsEnabled,
retentionDays,
cutoffUtc,
usageLogsDeleted: usageResult.deleted,
programLogsDeleted: programResult.deleted,
totalDeleted: usageResult.deleted + programResult.deleted,
};
}
+11 -10
View File
@@ -1,16 +1,12 @@
import { lt } from 'drizzle-orm';
import { config } from '../config.js';
import { db, schema } from '../db/index.js';
import { formatUtcSqlDateTime } from './localTimeService.js';
const DAY_MS = 24 * 60 * 60 * 1000;
import { cleanupUsageLogs, getLogCleanupCutoffUtc } from './logCleanupService.js';
let retentionTimer: ReturnType<typeof setInterval> | null = null;
export function getProxyLogRetentionCutoffUtc(nowMs = Date.now()): string | null {
const days = Math.max(0, Math.trunc(config.proxyLogRetentionDays));
if (days <= 0) return null;
return formatUtcSqlDateTime(new Date(nowMs - days * DAY_MS));
return getLogCleanupCutoffUtc(days, nowMs);
}
export async function cleanupExpiredProxyLogs(nowMs = Date.now()): Promise<{
@@ -30,10 +26,7 @@ export async function cleanupExpiredProxyLogs(nowMs = Date.now()): Promise<{
};
}
const deleted = (await db.delete(schema.proxyLogs)
.where(lt(schema.proxyLogs.createdAt, cutoffUtc))
.run())
.changes;
const { deleted } = await cleanupUsageLogs(retentionDays, nowMs);
return {
enabled: true,
@@ -68,3 +61,11 @@ export function stopProxyLogRetentionService(): void {
clearInterval(retentionTimer);
retentionTimer = null;
}
export function setLegacyProxyLogRetentionFallbackEnabled(enabled: boolean): void {
if (enabled) {
startProxyLogRetentionService();
return;
}
stopProxyLogRetentionService();
}
+122
View File
@@ -0,0 +1,122 @@
import { Headers, type HeadersInit } from 'undici';
export type SiteCustomHeadersRecord = Record<string, string>;
export type ParsedSiteCustomHeadersInput = {
present: boolean;
valid: boolean;
customHeaders: string | null;
headers: SiteCustomHeadersRecord | null;
error?: string;
};
function isPlainObject(value: unknown): value is Record<string, unknown> {
return Object.prototype.toString.call(value) === '[object Object]';
}
function normalizeSiteCustomHeadersRecord(input: Record<string, unknown>): SiteCustomHeadersRecord | null {
const normalized = new Headers();
for (const [rawKey, rawValue] of Object.entries(input)) {
const key = rawKey.trim();
if (!key) {
throw new Error('Header name cannot be empty.');
}
if (typeof rawValue !== 'string') {
throw new Error(`Header "${key}" must use a string value.`);
}
normalized.set(key, rawValue);
}
const entries = Array.from(normalized.entries()).sort(([left], [right]) => left.localeCompare(right));
if (entries.length === 0) {
return null;
}
return Object.fromEntries(entries);
}
export function parseSiteCustomHeadersInput(input: unknown): ParsedSiteCustomHeadersInput {
if (input === undefined) {
return { present: false, valid: true, customHeaders: null, headers: null };
}
if (input === null) {
return { present: true, valid: true, customHeaders: null, headers: null };
}
let parsedInput: unknown = input;
if (typeof input === 'string') {
const trimmed = input.trim();
if (!trimmed) {
return { present: true, valid: true, customHeaders: null, headers: null };
}
try {
parsedInput = JSON.parse(trimmed);
} catch {
return {
present: true,
valid: false,
customHeaders: null,
headers: null,
error: 'Invalid customHeaders. Expected a JSON object like {"x-header":"value"}.',
};
}
}
if (!isPlainObject(parsedInput)) {
return {
present: true,
valid: false,
customHeaders: null,
headers: null,
error: 'Invalid customHeaders. Expected a JSON object like {"x-header":"value"}.',
};
}
try {
const headers = normalizeSiteCustomHeadersRecord(parsedInput);
return {
present: true,
valid: true,
customHeaders: headers ? JSON.stringify(headers) : null,
headers,
};
} catch (error) {
return {
present: true,
valid: false,
customHeaders: null,
headers: null,
error: error instanceof Error
? `Invalid customHeaders. ${error.message}`
: 'Invalid customHeaders. Expected a JSON object like {"x-header":"value"}.',
};
}
}
export function readSiteCustomHeaders(input: unknown): SiteCustomHeadersRecord | null {
const parsed = parseSiteCustomHeadersInput(input);
if (!parsed.valid) {
return null;
}
return parsed.headers;
}
export function mergeHeadersWithSiteCustomHeaders(
siteCustomHeaders: unknown,
requestHeaders?: HeadersInit,
): HeadersInit | undefined {
const normalizedSiteHeaders = readSiteCustomHeaders(siteCustomHeaders);
if (!normalizedSiteHeaders) {
return requestHeaders;
}
const merged = new Headers(normalizedSiteHeaders);
if (requestHeaders) {
const explicitHeaders = new Headers(requestHeaders);
explicitHeaders.forEach((value, key) => {
merged.set(key, value);
});
}
return merged;
}
+46
View File
@@ -3,6 +3,7 @@ import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { sql } from 'drizzle-orm';
import { Headers } from 'undici';
type DbModule = typeof import('../db/index.js');
@@ -69,4 +70,49 @@ describe('siteProxy', () => {
expect('dispatcher' in requestInit).toBe(true);
});
it('merges site custom headers by matched request url and keeps explicit headers authoritative', async () => {
await db.insert(schema.sites).values({
name: 'headers-site',
url: 'https://headers-site.example.com',
platform: 'new-api',
customHeaders: JSON.stringify({
'cf-access-client-id': 'site-client',
authorization: 'Bearer site-default',
}),
}).run();
const { withSiteProxyRequestInit } = await import('./siteProxy.js');
const requestInit = await withSiteProxyRequestInit('https://headers-site.example.com/v1/models', {
method: 'GET',
headers: {
Authorization: 'Bearer request-token',
'X-Trace-Id': 'trace-1',
},
});
const headers = new Headers(requestInit.headers);
expect(headers.get('cf-access-client-id')).toBe('site-client');
expect(headers.get('authorization')).toBe('Bearer request-token');
expect(headers.get('x-trace-id')).toBe('trace-1');
});
it('merges site custom headers from site records even without cache lookup', async () => {
const { withSiteRecordProxyRequestInit } = await import('./siteProxy.js');
const requestInit = withSiteRecordProxyRequestInit({
useSystemProxy: false,
customHeaders: JSON.stringify({
'x-site-scope': 'site-level',
}),
}, {
method: 'POST',
headers: {
'X-Request-Id': 'req-1',
},
});
const headers = new Headers(requestInit.headers);
expect(headers.get('x-site-scope')).toBe('site-level');
expect(headers.get('x-request-id')).toBe('req-1');
});
});
+56 -16
View File
@@ -3,6 +3,7 @@ import { eq } from 'drizzle-orm';
import { config } from '../config.js';
import type { Dispatcher, RequestInit as UndiciRequestInit } from 'undici';
import { ProxyAgent } from 'undici';
import { mergeHeadersWithSiteCustomHeaders } from './siteCustomHeaders.js';
const SITE_PROXY_CACHE_TTL_MS = 3_000;
const SUPPORTED_PROXY_PROTOCOLS = new Set([
@@ -18,6 +19,7 @@ const SUPPORTED_PROXY_PROTOCOLS = new Set([
type SiteProxyRow = {
siteUrl: string;
useSystemProxy: boolean;
customHeaders: string | null;
};
type ParsedSiteProxyInput = {
@@ -28,6 +30,7 @@ type ParsedSiteProxyInput = {
type SiteProxyConfigLike = {
useSystemProxy?: boolean | null;
customHeaders?: string | null;
};
let siteProxyCache: {
@@ -67,6 +70,7 @@ async function getCachedSiteProxyRows(nowMs = Date.now()): Promise<SiteProxyRow[
.select({
siteUrl: schema.sites.url,
useSystemProxy: schema.sites.useSystemProxy,
customHeaders: schema.sites.customHeaders,
})
.from(schema.sites)
.all(),
@@ -92,6 +96,7 @@ async function getCachedSiteProxyRows(nowMs = Date.now()): Promise<SiteProxyRow[
rows: rows.map((row) => ({
siteUrl: normalizeSiteUrl(row.siteUrl),
useSystemProxy: !!row.useSystemProxy,
customHeaders: typeof row.customHeaders === 'string' ? row.customHeaders : null,
})),
systemProxyUrl: parsedSystemProxyUrl,
};
@@ -167,18 +172,11 @@ export function invalidateSiteProxyCache(): void {
siteProxyCache = { loadedAt: 0, rows: [], systemProxyUrl: null };
}
export async function resolveSiteProxyUrlByRequestUrl(requestUrl: string): Promise<string | null> {
const normalizedRequestUrl = normalizeSiteUrl(requestUrl);
if (!normalizedRequestUrl) return null;
const rows = await getCachedSiteProxyRows();
const systemProxyUrl = siteProxyCache.systemProxyUrl;
if (!systemProxyUrl) return null;
let bestMatch: string | null = null;
function findBestMatchingSiteRow(rows: SiteProxyRow[], normalizedRequestUrl: string): SiteProxyRow | null {
let bestMatch: SiteProxyRow | null = null;
let bestMatchLength = -1;
for (const row of rows) {
if (!row.useSystemProxy) continue;
if (!row.siteUrl) continue;
const isPrefixMatch = (
@@ -189,7 +187,7 @@ export async function resolveSiteProxyUrlByRequestUrl(requestUrl: string): Promi
if (!isPrefixMatch) continue;
if (row.siteUrl.length > bestMatchLength) {
bestMatch = systemProxyUrl;
bestMatch = row;
bestMatchLength = row.siteUrl.length;
}
}
@@ -197,18 +195,53 @@ export async function resolveSiteProxyUrlByRequestUrl(requestUrl: string): Promi
return bestMatch;
}
async function resolveSiteRequestConfigByRequestUrl(requestUrl: string): Promise<{
proxyUrl: string | null;
customHeaders: string | null;
}> {
const normalizedRequestUrl = normalizeSiteUrl(requestUrl);
if (!normalizedRequestUrl) {
return { proxyUrl: null, customHeaders: null };
}
const rows = await getCachedSiteProxyRows();
const matchedRow = findBestMatchingSiteRow(rows, normalizedRequestUrl);
const proxyUrl = matchedRow?.useSystemProxy ? siteProxyCache.systemProxyUrl : null;
return {
proxyUrl: proxyUrl || null,
customHeaders: matchedRow?.customHeaders ?? null,
};
}
export async function resolveSiteProxyUrlByRequestUrl(requestUrl: string): Promise<string | null> {
const resolved = await resolveSiteRequestConfigByRequestUrl(requestUrl);
return resolved.proxyUrl;
}
export async function withSiteProxyRequestInit(
requestUrl: string,
options?: UndiciRequestInit,
): Promise<UndiciRequestInit> {
const proxyUrl = await resolveSiteProxyUrlByRequestUrl(requestUrl);
if (!proxyUrl) return options ?? {};
const resolved = await resolveSiteRequestConfigByRequestUrl(requestUrl);
const nextOptions: UndiciRequestInit = {
...(options || {}),
};
const mergedHeaders = mergeHeadersWithSiteCustomHeaders(resolved.customHeaders, options?.headers);
if (mergedHeaders) {
nextOptions.headers = mergedHeaders;
}
const dispatcher = getDispatcherByProxyUrl(proxyUrl);
if (!dispatcher) return options ?? {};
if (!resolved.proxyUrl) {
return nextOptions;
}
const dispatcher = getDispatcherByProxyUrl(resolved.proxyUrl);
if (!dispatcher) {
return nextOptions;
}
return {
...(options || {}),
...nextOptions,
dispatcher,
};
}
@@ -238,5 +271,12 @@ export function withSiteRecordProxyRequestInit(
site: SiteProxyConfigLike | null | undefined,
options?: UndiciRequestInit,
): UndiciRequestInit {
return withExplicitProxyRequestInit(resolveProxyUrlForSite(site), options);
const nextOptions: UndiciRequestInit = {
...(options || {}),
};
const mergedHeaders = mergeHeadersWithSiteCustomHeaders(site?.customHeaders, options?.headers);
if (mergedHeaders) {
nextOptions.headers = mergedHeaders;
}
return withExplicitProxyRequestInit(resolveProxyUrlForSite(site), nextOptions);
}