diff --git a/job/compute-worker/package.json b/job/compute-worker/package.json index 0139e396e..1bc8d42e3 100644 --- a/job/compute-worker/package.json +++ b/job/compute-worker/package.json @@ -14,6 +14,7 @@ "dependencies": { "@constructive-io/job-pg": "^2.5.4", "@constructive-io/job-utils": "^2.5.4", + "@constructive-io/module-loader": "workspace:^", "@pgpmjs/logger": "^2.4.3", "pg": "8.20.0" }, diff --git a/job/compute-worker/src/billing.ts b/job/compute-worker/src/billing.ts index da23b3b6b..20752dea2 100644 --- a/job/compute-worker/src/billing.ts +++ b/job/compute-worker/src/billing.ts @@ -1,109 +1,13 @@ /** - * BillingTracker — quota checks and usage recording via the billing_module. - * - * Discovers billing configuration from metaschema_modules_public.billing_module. - * Gracefully no-ops when billing is not provisioned (standalone dev mode). + * BillingTracker — re-exports BillingLoader from @constructive-io/module-loader + * under the legacy name `BillingTracker`. */ -import { Logger } from '@pgpmjs/logger'; +import { BillingLoader } from '@constructive-io/module-loader'; import type { Pool } from 'pg'; -import { TtlCache } from './cache'; -import type { BillingModuleConfig } from './types'; - -const log = new Logger('compute:billing'); - -const BILLING_MODULE_SQL = ` - SELECT - s.schema_name AS public_schema, - ps.schema_name AS private_schema, - bm.record_usage_function - FROM metaschema_modules_public.billing_module bm - JOIN metaschema_public.schema s ON bm.schema_id = s.id - JOIN metaschema_public.schema ps ON bm.private_schema_id = ps.id - WHERE bm.database_id = $1 - LIMIT 1 -`; - -export class BillingTracker { - private pool: Pool; - private cache: TtlCache; - private databaseId: string; - +export class BillingTracker extends BillingLoader { constructor(pool: Pool, databaseId: string, cacheTtlMs?: number) { - this.pool = pool; - this.databaseId = databaseId; - this.cache = new TtlCache(cacheTtlMs ?? 60_000); - } - - async load(databaseId?: string): Promise { - const dbId = databaseId ?? this.databaseId; - const cached = this.cache.get(dbId); - if (cached !== undefined) return cached; - - try { - const { rows } = await this.pool.query(BILLING_MODULE_SQL, [dbId]); - if (!rows.length || !rows[0].record_usage_function) { - this.cache.set(dbId, null); - return null; - } - const config: BillingModuleConfig = { - publicSchema: rows[0].public_schema, - privateSchema: rows[0].private_schema, - recordUsageFunction: rows[0].record_usage_function, - }; - log.debug(`loaded billing module: ${config.privateSchema}.${config.recordUsageFunction}`); - this.cache.set(dbId, config); - return config; - } catch { - this.cache.set(dbId, null); - return null; - } - } - - /** - * Check if the entity has quota for this meter. - * Returns true if billing is not provisioned (graceful degradation). - */ - async checkQuota( - entityId: string, - meterSlug: string, - amount: number = 1, - databaseId?: string - ): Promise { - const config = await this.load(databaseId); - if (!config) return true; - - try { - const sql = `SELECT "${config.privateSchema}"."check_billing_quota"($1, $2::uuid, $3) AS allowed`; - const { rows } = await this.pool.query(sql, [meterSlug, entityId, amount]); - return rows[0]?.allowed !== false; - } catch (err: any) { - log.warn(`check_billing_quota failed (allowing): ${err.message}`); - return true; - } - } - - /** - * Record usage after a function completes. - * No-ops if billing is not provisioned. - */ - async recordUsage( - entityId: string, - meterSlug: string, - amount: number, - metadata: Record, - databaseId?: string - ): Promise { - const config = await this.load(databaseId); - if (!config) return; - - try { - const sql = `SELECT "${config.privateSchema}"."${config.recordUsageFunction}"($1, $2::uuid, $3, $4::jsonb)`; - await this.pool.query(sql, [meterSlug, entityId, amount, JSON.stringify(metadata)]); - log.debug(`recorded usage: ${meterSlug} entity=${entityId} amount=${amount}`); - } catch (err: any) { - log.warn(`record_usage failed (non-fatal): ${err.message}`); - } + super(pool, databaseId, cacheTtlMs); } } diff --git a/job/compute-worker/src/cache.ts b/job/compute-worker/src/cache.ts index 831ee457e..99d3c5221 100644 --- a/job/compute-worker/src/cache.ts +++ b/job/compute-worker/src/cache.ts @@ -1,44 +1,4 @@ /** - * Simple TTL cache — same pattern as agentic-server. - * Entries expire after the configured TTL. + * Re-export TtlCache from @constructive-io/module-loader. */ - -interface CacheEntry { - value: T; - expires_at: number; -} - -export class TtlCache { - private store = new Map>(); - private ttl_ms: number; - - constructor(ttl_ms: number) { - this.ttl_ms = ttl_ms; - } - - get(key: string): T | undefined { - const entry = this.store.get(key); - if (!entry) return undefined; - if (Date.now() > entry.expires_at) { - this.store.delete(key); - return undefined; - } - return entry.value; - } - - set(key: string, value: T): void { - this.store.set(key, { value, expires_at: Date.now() + this.ttl_ms }); - } - - delete(key: string): void { - this.store.delete(key); - } - - clear(): void { - this.store.clear(); - } - - get size(): number { - return this.store.size; - } -} +export { TtlCache } from '@constructive-io/module-loader'; diff --git a/job/compute-worker/src/compute-log.ts b/job/compute-worker/src/compute-log.ts index 9519d7b65..9fc7a6e19 100644 --- a/job/compute-worker/src/compute-log.ts +++ b/job/compute-worker/src/compute-log.ts @@ -6,11 +6,10 @@ * Gracefully no-ops if compute_log_module is not registered. */ +import type { ComputeModuleLoader } from '@constructive-io/module-loader'; import { Logger } from '@pgpmjs/logger'; import type { Pool } from 'pg'; -import { ComputeModuleLoader } from './module-loader'; - const log = new Logger('compute:log-tracker'); export interface ComputeLogEntry { diff --git a/job/compute-worker/src/discovery.ts b/job/compute-worker/src/discovery.ts index b27cc68c3..936bab0ad 100644 --- a/job/compute-worker/src/discovery.ts +++ b/job/compute-worker/src/discovery.ts @@ -8,11 +8,11 @@ * fetches the function definition and caches it for `ttlMs` (default 60 s). */ +import { TtlCache } from '@constructive-io/module-loader'; +import type { ComputeModuleLoader } from '@constructive-io/module-loader'; import { Logger } from '@pgpmjs/logger'; import type { Pool } from 'pg'; -import { TtlCache } from './cache'; -import type { ComputeModuleLoader } from './module-loader'; import type { PlatformFunctionDefinition } from './types'; const log = new Logger('compute:discovery'); diff --git a/job/compute-worker/src/index.ts b/job/compute-worker/src/index.ts index 17fb6cc2e..a6ef430c0 100644 --- a/job/compute-worker/src/index.ts +++ b/job/compute-worker/src/index.ts @@ -15,6 +15,8 @@ import poolManager from '@constructive-io/job-pg'; import type { PgClientLike } from '@constructive-io/job-utils'; import * as jobs from '@constructive-io/job-utils'; +import { ComputeModuleLoader } from '@constructive-io/module-loader'; +import type { GraphExecutionModuleConfig } from '@constructive-io/module-loader'; import { Logger } from '@pgpmjs/logger'; import type { Pool, PoolClient } from 'pg'; @@ -23,7 +25,6 @@ import { ComputeLogTracker } from './compute-log'; import { FunctionDiscovery } from './discovery'; import { executeInline, getInlineImpl } from './inline'; import { InvocationTracker } from './invocation'; -import { ComputeModuleLoader } from './module-loader'; import { compute_request } from './req'; import type { ComputeJobRow, ComputeWorkerOptions, PlatformFunctionDefinition } from './types'; import { isGraphNodePayload } from './types'; @@ -53,6 +54,7 @@ export type { FunctionPortDefinition, FunctionRequirement, FunctionRuntime, + GraphExecutionModuleConfig, GraphNodePayload, InvocationModuleConfig, InvocationStatus, @@ -622,6 +624,14 @@ export default class ComputeWorker { } } + /** + * Resolve graph execution module config (cached via ComputeModuleLoader). + */ + private async graphConfig(): Promise { + const config = await this.loader.load(this.platformDatabaseId); + return config.graphExecutionModule; + } + /** * Transition a node from queued → running when the worker picks up the job. */ @@ -630,8 +640,9 @@ export default class ComputeWorker { nodeName: string ): Promise { log.debug('marking graph node running', { executionId, nodeName }); + const ge = await this.graphConfig(); await this.pgPool.query( - `UPDATE constructive_compute_public.platform_function_graph_execution_node_states + `UPDATE "${ge.publicSchema}"."${ge.nodeStatesTable}" SET status = 'running', started_at = now() WHERE execution_id = $1::uuid AND node_name = $2 AND status = 'queued'`, [executionId, nodeName] @@ -649,8 +660,9 @@ export default class ComputeWorker { output: unknown ): Promise { log.debug('completing graph node', { executionId, nodeName }); + const ge = await this.graphConfig(); await this.pgPool.query( - `SELECT constructive_compute_private.platform_complete_node($1::uuid, $2, $3::jsonb)`, + `SELECT "${ge.privateSchema}"."${ge.completeNodeFunction}"($1::uuid, $2, $3::jsonb)`, [executionId, nodeName, JSON.stringify(output ?? {})] ); } @@ -667,13 +679,12 @@ export default class ComputeWorker { ): Promise { log.error('graph node failed', { executionId, nodeName, error: errorMessage }); try { + const ge = await this.graphConfig(); await this.pgPool.query( - `SELECT constructive_compute_private.platform_fail_node($1::uuid, $2, $3, $4)`, + `SELECT "${ge.privateSchema}"."${ge.failNodeFunction}"($1::uuid, $2, $3, $4)`, [executionId, nodeName, 'NODE_EXECUTION_FAILED', errorMessage] ); } catch (err: any) { - // Execution may already be completed/failed (race with graphOutput). - // Log so the error is never invisible. log.warn('platform_fail_node raised; execution may already be finished', { executionId, nodeName, error: errorMessage, sqlError: err.message, }); diff --git a/job/compute-worker/src/invocation.ts b/job/compute-worker/src/invocation.ts index 09d675fe5..8e660ef5e 100644 --- a/job/compute-worker/src/invocation.ts +++ b/job/compute-worker/src/invocation.ts @@ -14,11 +14,11 @@ * 3. `fail()` — updates to status='failed' with error + duration */ +import type { ComputeModuleLoader, InvocationModuleConfig } from '@constructive-io/module-loader'; import { Logger } from '@pgpmjs/logger'; import type { Pool } from 'pg'; -import type { ComputeModuleLoader } from './module-loader'; -import type { CreateInvocationInput, InvocationModuleConfig } from './types'; +import type { CreateInvocationInput } from './types'; const log = new Logger('compute:invocation'); diff --git a/job/compute-worker/src/module-loader.ts b/job/compute-worker/src/module-loader.ts index 9fe69fd1b..4236122a6 100644 --- a/job/compute-worker/src/module-loader.ts +++ b/job/compute-worker/src/module-loader.ts @@ -1,142 +1,5 @@ /** - * ComputeModuleLoader — resolves compute module schema/table names - * dynamically from metaschema instead of hardcoding schema references. - * - * Queries metaschema_modules_public.function_module and - * metaschema_modules_public.function_invocation_module, joined with - * metaschema_public.schema to resolve schema names. Results are - * TTL-cached per database_id. + * Re-export ComputeModuleLoader from @constructive-io/module-loader. */ - -import { Logger } from '@pgpmjs/logger'; -import type { Pool } from 'pg'; - -import { TtlCache } from './cache'; -import type { ComputeLogModuleConfig, ComputeModuleConfig, FunctionModuleConfig, InvocationModuleConfig } from './types'; - -const log = new Logger('compute:module-loader'); - -const FUNCTION_MODULE_SQL = ` - SELECT - s.schema_name AS public_schema, - ps.schema_name AS private_schema, - fm.definitions_table_name, - fm.secret_definitions_table_name, - fm.scope - FROM metaschema_modules_public.function_module fm - JOIN metaschema_public.schema s ON fm.schema_id = s.id - JOIN metaschema_public.schema ps ON fm.private_schema_id = ps.id - WHERE fm.database_id = $1 -`; - -const INVOCATION_MODULE_SQL = ` - SELECT - s.schema_name AS public_schema, - fim.invocations_table_name, - fim.execution_logs_table_name, - fim.scope - FROM metaschema_modules_public.function_invocation_module fim - JOIN metaschema_public.schema s ON fim.schema_id = s.id - WHERE fim.database_id = $1 -`; - -const COMPUTE_LOG_MODULE_SQL = ` - SELECT - s.schema_name AS public_schema, - ps.schema_name AS private_schema, - clm.compute_log_table_name, - clm.usage_daily_table_name, - clm.scope - FROM metaschema_modules_public.compute_log_module clm - JOIN metaschema_public.schema s ON clm.schema_id = s.id - JOIN metaschema_public.schema ps ON clm.private_schema_id = ps.id - WHERE clm.database_id = $1 -`; - -export class ComputeModuleLoader { - private cache: TtlCache; - private pool: Pool; - - constructor(pool: Pool, ttlMs = 60_000) { - this.pool = pool; - this.cache = new TtlCache(ttlMs); - } - - async load(databaseId: string): Promise { - const cached = this.cache.get(databaseId); - if (cached !== undefined) { - log.debug(`module config cache hit for database ${databaseId}`); - return cached; - } - - log.debug(`module config cache miss for database ${databaseId}, querying metaschema`); - - const fnResult = await this.pool.query(FUNCTION_MODULE_SQL, [databaseId]); - - let functionModule: FunctionModuleConfig | null = null; - if (fnResult.rows.length > 0) { - const row = fnResult.rows[0]; - functionModule = { - publicSchema: row.public_schema, - privateSchema: row.private_schema, - definitionsTable: row.definitions_table_name, - secretDefinitionsTable: row.secret_definitions_table_name, - scope: row.scope, - }; - } - - // Invocation module is optional — the table may not be deployed yet - let invocationModules: InvocationModuleConfig[] = []; - try { - const invResult = await this.pool.query(INVOCATION_MODULE_SQL, [databaseId]); - invocationModules = invResult.rows.map( - (row: Record) => ({ - publicSchema: row.public_schema, - invocationsTable: row.invocations_table_name, - executionLogsTable: row.execution_logs_table_name, - scope: row.scope, - }) - ); - } catch { - log.debug(`function_invocation_module not available for database ${databaseId} — invocation tracking disabled`); - } - - // Compute log module is optional — only present when usage tracking is provisioned - let computeLogModule: ComputeLogModuleConfig | null = null; - try { - const clResult = await this.pool.query(COMPUTE_LOG_MODULE_SQL, [databaseId]); - if (clResult.rows.length > 0) { - const row = clResult.rows[0]; - computeLogModule = { - publicSchema: row.public_schema, - privateSchema: row.private_schema, - computeLogTable: row.compute_log_table_name, - usageDailyTable: row.usage_daily_table_name, - scope: row.scope, - }; - } - } catch { - log.debug(`compute_log_module not available for database ${databaseId} — usage logging disabled`); - } - - const config: ComputeModuleConfig = { functionModule, invocationModules, computeLogModule }; - this.cache.set(databaseId, config); - - log.info( - `loaded compute module config for database ${databaseId}: ` + - `fn=${functionModule ? `${functionModule.publicSchema}.${functionModule.definitionsTable}` : 'none'}, ` + - `invocations=${invocationModules.length} scope(s), ` + - `computeLog=${computeLogModule ? `${computeLogModule.publicSchema}.${computeLogModule.computeLogTable}` : 'none'}` - ); - - return config; - } - - invalidate(databaseId: string): void { - this.cache.delete(databaseId); - } - - invalidateAll(): void { - this.cache.clear(); - } -} +export { ComputeModuleLoader } from '@constructive-io/module-loader'; +export type { ComputeModuleConfigExtended } from '@constructive-io/module-loader'; diff --git a/job/compute-worker/src/types.ts b/job/compute-worker/src/types.ts index a837e6f2d..b62cf5ddb 100644 --- a/job/compute-worker/src/types.ts +++ b/job/compute-worker/src/types.ts @@ -1,5 +1,15 @@ import type { Pool } from 'pg'; +// Re-export module config types from the shared package +export type { + BillingModuleConfig, + ComputeLogModuleConfig, + ComputeModuleConfig, + FunctionModuleConfig, + GraphExecutionModuleConfig, + InvocationModuleConfig, +} from '@constructive-io/module-loader'; + // ─── Platform Function Definition ──────────────────────────────────────────── export interface FunctionRequirement { @@ -36,37 +46,6 @@ export interface PlatformFunctionDefinition { outputs: FunctionPortDefinition[] | null; } -// ─── Compute Module Config ──────────────────────────────────────────────────── - -export interface FunctionModuleConfig { - publicSchema: string; - privateSchema: string; - definitionsTable: string; - secretDefinitionsTable: string; - scope: string; -} - -export interface InvocationModuleConfig { - publicSchema: string; - invocationsTable: string; - executionLogsTable: string; - scope: string; -} - -export interface ComputeLogModuleConfig { - publicSchema: string; - privateSchema: string; - computeLogTable: string; - usageDailyTable: string; - scope: string; -} - -export interface ComputeModuleConfig { - functionModule: FunctionModuleConfig | null; - invocationModules: InvocationModuleConfig[]; - computeLogModule: ComputeLogModuleConfig | null; -} - // ─── Invocation Record ─────────────────────────────────────────────────────── export type InvocationStatus = 'running' | 'completed' | 'failed'; @@ -108,10 +87,6 @@ export interface ComputeJobRow { // ─── Graph Execution ───────────────────────────────────────────────────────── -/** - * When a job is enqueued by tick_execution for a graph node, - * its payload includes these fields alongside the node's assembled inputs. - */ export interface GraphNodePayload { execution_id: string; node_name: string; @@ -121,9 +96,6 @@ export interface GraphNodePayload { node_path?: string[]; } -/** - * Type guard: does this payload represent a graph node dispatch? - */ export function isGraphNodePayload(payload: unknown): payload is GraphNodePayload { if (!payload || typeof payload !== 'object') return false; const p = payload as Record; @@ -135,14 +107,8 @@ export function isGraphNodePayload(payload: unknown): payload is GraphNodePayloa // ─── Billing ────────────────────────────────────────────────────────────────── -export interface BillingModuleConfig { - publicSchema: string; - privateSchema: string; - recordUsageFunction: string; -} - export interface BillingContext { - config: BillingModuleConfig; + config: import('@constructive-io/module-loader').BillingModuleConfig; entityId: string; meterSlug: string; } diff --git a/job/worker/package.json b/job/worker/package.json index 658f5f97b..5e9943a59 100644 --- a/job/worker/package.json +++ b/job/worker/package.json @@ -18,6 +18,8 @@ "dependencies": { "@constructive-io/job-pg": "^2.5.4", "@constructive-io/job-utils": "^2.5.4", + "@constructive-io/module-loader": "workspace:^", + "@constructive-io/usage-loader": "workspace:^", "@pgpmjs/logger": "^2.4.3", "pg": "8.20.0" }, diff --git a/job/worker/src/compute-meter.ts b/job/worker/src/compute-meter.ts index fb5143c08..0f0a3c468 100644 --- a/job/worker/src/compute-meter.ts +++ b/job/worker/src/compute-meter.ts @@ -1,36 +1,35 @@ /** * Compute metering — fire-and-forget usage logging for every job. * - * Delegates to UsageClient which resolves table names dynamically from - * MetaSchema module registration tables. Schema-qualified references - * (invocations table + usage log) are discovered at runtime, not hardcoded. + * Delegates to @constructive-io/usage-loader which resolves table names + * dynamically from MetaSchema module registration tables. * * All writes are non-blocking: errors are logged and swallowed so * metering never affects job throughput or latency. */ import type { Pool } from 'pg'; - -import type { MeterEntry } from './usage-client'; -import { UsageClient } from './usage-client'; +import { UsageLoader } from '@constructive-io/usage-loader'; +import type { MeterEntry } from '@constructive-io/usage-loader'; export type { MeterEntry }; -/** Module-level client cache — reused across calls within the same pool. */ -let _client: UsageClient | null = null; +/** Module-level loader cache — reused across calls within the same pool. */ +let _loader: UsageLoader | null = null; let _pool: Pool | null = null; -function getOrCreateClient(pool: Pool): UsageClient { - if (_client && _pool === pool) return _client; - _client = new UsageClient(pool); +function getOrCreateLoader(pool: Pool): UsageLoader { + if (_loader && _pool === pool) return _loader; + _loader = new UsageLoader(pool); _pool = pool; - return _client; + return _loader; } /** * Log a compute invocation to both the invocations and usage tables. * Fire-and-forget: returns immediately, never throws. + * Returns the generated invocation_id (UUID). */ -export function logComputeUsage(pool: Pool, entry: MeterEntry): void { - getOrCreateClient(pool).logComputeUsage(entry); +export function logComputeUsage(pool: Pool, entry: MeterEntry): string { + return getOrCreateLoader(pool).logComputeUsage(entry); } diff --git a/job/worker/src/graph-complete.ts b/job/worker/src/graph-complete.ts index 67faa6455..466d6c122 100644 --- a/job/worker/src/graph-complete.ts +++ b/job/worker/src/graph-complete.ts @@ -5,37 +5,48 @@ * call the corresponding SQL procedures to store outputs and advance * the execution graph to the next tick. * - * This is the shared completion path for BOTH inline (FBP native) and - * cloud function invocations — the graph engine sees no difference. + * Schema/function names are resolved dynamically via ModuleLoader + * (graph_execution_module) instead of hardcoding. */ +import { ComputeModuleLoader, DEFAULT_DATABASE_ID } from '@constructive-io/module-loader'; +import type { GraphExecutionModuleConfig } from '@constructive-io/module-loader'; import type { Pool } from 'pg'; +let _loader: ComputeModuleLoader | null = null; +let _pool: Pool | null = null; + +function getLoader(pool: Pool): ComputeModuleLoader { + if (_loader && _pool === pool) return _loader; + _loader = new ComputeModuleLoader(pool); + _pool = pool; + return _loader; +} + +async function resolveGraph(pool: Pool): Promise { + const config = await getLoader(pool).load(DEFAULT_DATABASE_ID); + return config.graphExecutionModule; +} + /** * Mark a graph node as completed with its output data. - * Calls `constructive_compute_private.platform_complete_node` which: - * 1. Stores the output in execution_outputs (content-addressed) - * 2. Updates node_outputs on the execution - * 3. Sets node_state to 'completed' - * 4. Calls tick_execution to advance the graph + * Resolves the complete_node function from MetaSchema. */ export const completeNode = async ( pool: Pool, executionId: string, nodeName: string, - outputData: Record + outputData: Record ): Promise => { + const ge = await resolveGraph(pool); await pool.query( - `SELECT "constructive_compute_private".platform_complete_node($1::uuid, $2::text, $3::jsonb)`, + `SELECT "${ge.privateSchema}"."${ge.completeNodeFunction}"($1::uuid, $2::text, $3::jsonb)`, [executionId, nodeName, JSON.stringify(outputData)] ); }; /** * Mark a graph node as failed with an error message. - * Calls `constructive_compute_private.platform_fail_node` which: - * 1. Sets node_state to 'failed' - * 2. Stores the error message - * 3. Optionally fails the parent execution + * Resolves the fail_node function from MetaSchema. */ export const failNode = async ( pool: Pool, @@ -43,8 +54,15 @@ export const failNode = async ( nodeName: string, errorMessage: string ): Promise => { + const ge = await resolveGraph(pool); await pool.query( - `SELECT "constructive_compute_private".platform_fail_node($1::uuid, $2::text, $3::text)`, + `SELECT "${ge.privateSchema}"."${ge.failNodeFunction}"($1::uuid, $2::text, $3::text)`, [executionId, nodeName, errorMessage] ); }; + +/** Reset loader cache (for testing). */ +export function _resetGraphCompleteCache(): void { + _loader = null; + _pool = null; +} diff --git a/job/worker/src/index.ts b/job/worker/src/index.ts index 68fa1e7b6..1f6918439 100644 --- a/job/worker/src/index.ts +++ b/job/worker/src/index.ts @@ -353,5 +353,5 @@ export default class Worker { } export { Worker }; -export type { InferenceEntry, MeterEntry, UsageModuleConfig } from './usage-client'; -export { createUsageClient,UsageClient } from './usage-client'; +export type { InferenceEntry, MeterEntry, StorageEntry, UsageModuleConfig, UsageTableConfig } from './usage-client'; +export { UsageClient, UsageLoader, getLoader, _resetLoaderCache } from './usage-client'; diff --git a/job/worker/src/storage-meter.ts b/job/worker/src/storage-meter.ts index 2987c519a..2f21155fa 100644 --- a/job/worker/src/storage-meter.ts +++ b/job/worker/src/storage-meter.ts @@ -1,28 +1,28 @@ /** * Storage metering — fire-and-forget usage logging for S3/MinIO operations. * - * Logs to `constructive_usage_public.platform_usage_log_storage` - * after each read/write/delete against object storage. + * Delegates to @constructive-io/usage-loader which resolves table names + * dynamically from MetaSchema module registration tables. * * All writes are non-blocking: errors are logged and swallowed so * metering never affects function execution or storage latency. */ -import { Logger } from '@pgpmjs/logger'; import type { Pool } from 'pg'; -import { randomUUID } from 'crypto'; +import { UsageLoader } from '@constructive-io/usage-loader'; +import type { StorageEntry } from '@constructive-io/usage-loader'; -const log = new Logger('storage-meter'); +export type { StorageEntry }; -export interface StorageEntry { - databaseId?: string; - entityId?: string; - actorId?: string; - operation: 'read' | 'write' | 'delete'; - bucket: string; - key: string; - sizeBytes: number; - durationMs: number; +/** Module-level loader cache — reused across calls within the same pool. */ +let _loader: UsageLoader | null = null; +let _pool: Pool | null = null; + +function getOrCreateLoader(pool: Pool): UsageLoader { + if (_loader && _pool === pool) return _loader; + _loader = new UsageLoader(pool); + _pool = pool; + return _loader; } /** @@ -30,29 +30,5 @@ export interface StorageEntry { * Fire-and-forget: returns immediately, never throws. */ export function logStorageUsage(pool: Pool, entry: StorageEntry): void { - const id = randomUUID(); - const now = new Date(); - - pool - .query( - `INSERT INTO "constructive_usage_public".platform_usage_log_storage - (id, database_id, entity_id, actor_id, operation, - bucket, key, size_bytes, duration_ms, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, - [ - id, - entry.databaseId ?? null, - entry.entityId ?? null, - entry.actorId ?? null, - entry.operation, - entry.bucket, - entry.key, - entry.sizeBytes, - Math.round(entry.durationMs), - now - ] - ) - .catch((err) => { - log.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); + getOrCreateLoader(pool).logStorageUsage(entry); } diff --git a/job/worker/src/usage-client.ts b/job/worker/src/usage-client.ts index f2c98e80a..4e9cf5a35 100644 --- a/job/worker/src/usage-client.ts +++ b/job/worker/src/usage-client.ts @@ -1,295 +1,22 @@ /** - * UsageClient — typed, fire-and-forget usage logging for compute and inference. + * Re-export from @constructive-io/usage-loader for backward compatibility. * - * Resolves table names dynamically from MetaSchema module registration tables - * (metaschema_public.schema + metaschema_public.table) rather than hardcoding - * schema-qualified references. Results are TTL-cached per client instance. - * - * When MetaSchema is unavailable (tests, bootstrapping) the client falls back - * to the well-known default table names so logging still works. + * The canonical implementation is in packages/usage-loader. + * This file exists so existing imports from within job/worker/ continue working. */ -import { Logger } from '@pgpmjs/logger'; -import { randomUUID } from 'crypto'; -import type { Pool } from 'pg'; - -const log = new Logger('usage-client'); - -// ─── Module Resolution ──────────────────────────────────────────────────────── - -/** - * Query to discover usage table locations from metaschema. - * Joins metaschema_public.table → metaschema_public.schema to resolve - * the schema_name for each registered table. - */ -const USAGE_TABLES_SQL = ` - SELECT s.schema_name, t.name AS table_name - FROM metaschema_public."table" t - JOIN metaschema_public.schema s ON t.schema_id = s.id - WHERE t.database_id = $1 - AND t.name = ANY($2) -`; - -/** Well-known table names we look up in metaschema */ -const KNOWN_TABLES = [ - 'platform_function_invocations', - 'platform_usage_log_computes', - 'platform_usage_log_inferences', -] as const; - -/** Default config used when metaschema is unavailable */ -const DEFAULTS: UsageModuleConfig = { - invocationsSchema: 'constructive_compute_public', - invocationsTable: 'platform_function_invocations', - computeUsageSchema: 'constructive_usage_public', - computeUsageTable: 'platform_usage_log_computes', - inferenceUsageSchema: 'constructive_usage_public', - inferenceUsageTable: 'platform_usage_log_inferences', -}; - -export interface UsageModuleConfig { - invocationsSchema: string; - invocationsTable: string; - computeUsageSchema: string; - computeUsageTable: string; - inferenceUsageSchema: string; - inferenceUsageTable: string; -} - -// ─── Entry Types ────────────────────────────────────────────────────────────── - -export interface MeterEntry { - jobId: number | string; - taskIdentifier: string; - databaseId?: string; - actorId?: string; - entityId?: string; - durationMs: number; - status: 'ok' | 'error'; - error?: string; - payload?: unknown; - result?: unknown; - graphExecutionId?: string; - nodeName?: string; - dispatchType: 'inline' | 'http'; -} - -export interface InferenceEntry { - databaseId?: string; - entityId?: string; - actorId?: string; - requestId?: string; - model: string; - provider: string; - service: 'chat' | 'embed'; - operation: string; - inputTokens: number; - outputTokens: number; - totalTokens: number; - latencyMs: number; - status: 'ok' | 'error'; - errorType?: string; - rawUsage?: unknown; -} - -// ─── UsageClient ────────────────────────────────────────────────────────────── - -const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000'; -const CONFIG_TTL_MS = 60_000; - -export class UsageClient { - private pool: Pool; - private databaseId: string; - private config: UsageModuleConfig | null = null; - private configPromise: Promise | null = null; - private configExpiresAt = 0; - - constructor(pool: Pool, databaseId: string = DEFAULT_DATABASE_ID) { - this.pool = pool; - this.databaseId = databaseId; - } - - // ─── Config Resolution ──────────────────────────────────────────────── - - private resolveConfig(): Promise { - if (this.config && Date.now() < this.configExpiresAt) { - return Promise.resolve(this.config); - } - if (this.configPromise) return this.configPromise; - this.configPromise = this.loadConfig(); - return this.configPromise; - } - - private async loadConfig(): Promise { - try { - const { rows } = await this.pool.query(USAGE_TABLES_SQL, [ - this.databaseId, - [...KNOWN_TABLES], - ]); - - const config: UsageModuleConfig = { ...DEFAULTS }; - for (const row of rows) { - if (row.table_name === 'platform_function_invocations') { - config.invocationsSchema = row.schema_name; - } - if (row.table_name === 'platform_usage_log_computes') { - config.computeUsageSchema = row.schema_name; - } - if (row.table_name === 'platform_usage_log_inferences') { - config.inferenceUsageSchema = row.schema_name; - } - } - - this.config = config; - this.configExpiresAt = Date.now() + CONFIG_TTL_MS; - this.configPromise = null; - - log.debug( - `resolved usage config: invocations=${config.invocationsSchema}.${config.invocationsTable}, ` + - `compute=${config.computeUsageSchema}.${config.computeUsageTable}, ` + - `inference=${config.inferenceUsageSchema}.${config.inferenceUsageTable}` - ); - - return config; - } catch { - log.debug('metaschema lookup unavailable — using default table names'); - this.config = DEFAULTS; - this.configExpiresAt = Date.now() + CONFIG_TTL_MS; - this.configPromise = null; - return DEFAULTS; - } - } - - // ─── Compute Usage ──────────────────────────────────────────────────── - - /** - * Log a compute invocation to both the invocations and usage tables. - * Fire-and-forget: returns immediately, never throws. - */ - logComputeUsage(entry: MeterEntry): void { - this.resolveConfig() - .then((cfg) => this.insertComputeUsage(cfg, entry)) - .catch((err) => { - log.warn(`compute usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - } - - private async insertComputeUsage(cfg: UsageModuleConfig, entry: MeterEntry): Promise { - const invocationId = randomUUID(); - const now = new Date(); - const startedAt = new Date(now.getTime() - entry.durationMs); - - // INSERT into invocations table - this.pool - .query( - `INSERT INTO "${cfg.invocationsSchema}"."${cfg.invocationsTable}" - (id, database_id, actor_id, task_identifier, job_id, - graph_execution_id, status, duration_ms, - started_at, completed_at, payload, result, error, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)`, - [ - invocationId, - entry.databaseId ?? null, - entry.actorId ?? null, - entry.taskIdentifier, - typeof entry.jobId === 'string' ? parseInt(entry.jobId, 10) || 0 : entry.jobId, - entry.graphExecutionId ?? null, - entry.status, - Math.round(entry.durationMs), - startedAt, - now, - entry.payload ? JSON.stringify(entry.payload) : null, - entry.result ? JSON.stringify(entry.result) : null, - entry.error ?? null, - now, - ] - ) - .catch((err) => { - log.warn(`invocation log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - - // INSERT into usage log table - this.pool - .query( - `INSERT INTO "${cfg.computeUsageSchema}"."${cfg.computeUsageTable}" - (id, database_id, entity_id, actor_id, task_identifier, - job_id, invocation_id, status, duration_ms, error, completed_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`, - [ - randomUUID(), - entry.databaseId ?? null, - entry.entityId ?? null, - entry.actorId ?? null, - entry.taskIdentifier, - typeof entry.jobId === 'string' ? parseInt(entry.jobId, 10) || 0 : entry.jobId, - invocationId, - entry.status, - Math.round(entry.durationMs), - entry.error ?? null, - now, - ] - ) - .catch((err) => { - log.warn(`usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - } - - // ─── Inference Usage ────────────────────────────────────────────────── - - /** - * Log an inference invocation to the usage log table. - * Fire-and-forget: returns immediately, never throws. - */ - logInferenceUsage(entry: InferenceEntry): void { - this.resolveConfig() - .then((cfg) => this.insertInferenceUsage(cfg, entry)) - .catch((err) => { - log.warn(`inference usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - } - - private async insertInferenceUsage(cfg: UsageModuleConfig, entry: InferenceEntry): Promise { - const id = randomUUID(); - const now = new Date(); - - this.pool - .query( - `INSERT INTO "${cfg.inferenceUsageSchema}"."${cfg.inferenceUsageTable}" - (id, database_id, entity_id, actor_id, request_id, - model, provider, service, operation, - input_tokens, output_tokens, total_tokens, - latency_ms, status, error_type, raw_usage, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`, - [ - id, - entry.databaseId ?? null, - entry.entityId ?? null, - entry.actorId ?? null, - entry.requestId ?? randomUUID(), - entry.model, - entry.provider, - entry.service, - entry.operation, - entry.inputTokens, - entry.outputTokens, - entry.totalTokens, - Math.round(entry.latencyMs), - entry.status, - entry.errorType ?? null, - entry.rawUsage ? JSON.stringify(entry.rawUsage) : null, - now, - ] - ) - .catch((err) => { - log.warn(`inference log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - } -} - -/** - * Create a UsageClient instance. - * Convenience factory for callers that prefer a functional style. - */ -export function createUsageClient(pool: Pool, databaseId?: string): UsageClient { - return new UsageClient(pool, databaseId); -} +export { + UsageLoader as UsageClient, + UsageLoader, + getLoader, + _resetLoaderCache, + DEFAULTS, +} from '@constructive-io/usage-loader'; + +export type { + UsageTableConfig as UsageModuleConfig, + UsageTableConfig, + MeterEntry, + InferenceEntry, + StorageEntry, +} from '@constructive-io/usage-loader'; diff --git a/packages/agentic-server/package.json b/packages/agentic-server/package.json index 7f65f70fd..6e93891d9 100644 --- a/packages/agentic-server/package.json +++ b/packages/agentic-server/package.json @@ -19,6 +19,7 @@ "start": "node dist/standalone.js" }, "dependencies": { + "@constructive-io/usage-loader": "workspace:^", "@pgpmjs/logger": "^2.4.3", "express": "5.2.1" }, diff --git a/packages/agentic-server/src/index.ts b/packages/agentic-server/src/index.ts index c97f6e4b7..5a768b3d3 100644 --- a/packages/agentic-server/src/index.ts +++ b/packages/agentic-server/src/index.ts @@ -15,4 +15,4 @@ export { createAgenticServer } from './server'; export { createRouter } from './router'; export { logInferenceUsage } from './inference-meter'; export type { InferenceEntry } from './inference-meter'; -export type { AgenticRouterOptions } from './router'; +export type { AgenticRouterOptions, ProviderConfig } from './router'; diff --git a/packages/agentic-server/src/inference-meter.ts b/packages/agentic-server/src/inference-meter.ts index 83027c7eb..32ef19d30 100644 --- a/packages/agentic-server/src/inference-meter.ts +++ b/packages/agentic-server/src/inference-meter.ts @@ -1,147 +1,40 @@ /** * Inference metering — fire-and-forget usage logging for LLM calls. * - * Self-contained implementation that resolves table names dynamically - * from MetaSchema module registration tables. Falls back to well-known - * defaults when MetaSchema is unavailable. + * Delegates to @constructive-io/usage-loader which resolves table names + * dynamically from MetaSchema module registration tables. * * All writes are non-blocking: errors are logged and swallowed so * metering never affects inference latency or response delivery. */ -import { Logger } from '@pgpmjs/logger'; -import { randomUUID } from 'crypto'; import type { Pool } from 'pg'; +import { UsageLoader, getLoader, _resetLoaderCache } from '@constructive-io/usage-loader'; +import type { InferenceEntry } from '@constructive-io/usage-loader'; -const log = new Logger('inference-meter'); +export type { InferenceEntry }; -// ─── Types ──────────────────────────────────────────────────────────────────── +/** Module-level loader cache — reused across calls within the same pool. */ +let _loader: UsageLoader | null = null; +let _pool: Pool | null = null; -export interface InferenceEntry { - databaseId?: string; - entityId?: string; - actorId?: string; - requestId?: string; - model: string; - provider: string; - service: 'chat' | 'embed'; - operation: string; - inputTokens: number; - outputTokens: number; - totalTokens: number; - latencyMs: number; - status: 'ok' | 'error'; - errorType?: string; - rawUsage?: unknown; +function getOrCreateLoader(pool: Pool): UsageLoader { + if (_loader && _pool === pool) return _loader; + _loader = new UsageLoader(pool); + _pool = pool; + return _loader; } -// ─── Table Resolution ───────────────────────────────────────────────────────── - -const INFERENCE_TABLE_SQL = ` - SELECT s.schema_name, t.name AS table_name - FROM metaschema_public."table" t - JOIN metaschema_public.schema s ON t.schema_id = s.id - WHERE t.database_id = $1 - AND t.name = 'platform_usage_log_inferences' - LIMIT 1 -`; - -const DEFAULT_SCHEMA = 'constructive_usage_public'; -const DEFAULT_TABLE = 'platform_usage_log_inferences'; -const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000'; -const CONFIG_TTL_MS = 60_000; - -interface TableRef { - schema: string; - table: string; -} - -let _cachedRef: TableRef | null = null; -let _cacheExpiresAt = 0; -let _resolvePromise: Promise | null = null; - -/** Reset the module-level cache (for testing). */ -export function _resetCache(): void { - _cachedRef = null; - _cacheExpiresAt = 0; - _resolvePromise = null; -} - -async function resolveTable(pool: Pool, databaseId: string): Promise { - if (_cachedRef && Date.now() < _cacheExpiresAt) return _cachedRef; - if (_resolvePromise) return _resolvePromise; - - _resolvePromise = (async () => { - try { - const { rows } = await pool.query(INFERENCE_TABLE_SQL, [databaseId]); - const ref: TableRef = - rows.length > 0 - ? { schema: rows[0].schema_name, table: rows[0].table_name } - : { schema: DEFAULT_SCHEMA, table: DEFAULT_TABLE }; - _cachedRef = ref; - _cacheExpiresAt = Date.now() + CONFIG_TTL_MS; - _resolvePromise = null; - return ref; - } catch { - log.debug('metaschema lookup unavailable — using default table names'); - const ref: TableRef = { schema: DEFAULT_SCHEMA, table: DEFAULT_TABLE }; - _cachedRef = ref; - _cacheExpiresAt = Date.now() + CONFIG_TTL_MS; - _resolvePromise = null; - return ref; - } - })(); - - return _resolvePromise; -} - -// ─── Public API ─────────────────────────────────────────────────────────────── - /** * Log an inference invocation to the usage log table. * Fire-and-forget: returns immediately, never throws. */ export function logInferenceUsage(pool: Pool, entry: InferenceEntry): void { - const dbId = entry.databaseId ?? DEFAULT_DATABASE_ID; - - resolveTable(pool, dbId) - .then((ref) => { - const id = randomUUID(); - const now = new Date(); + getOrCreateLoader(pool).logInferenceUsage(entry); +} - pool - .query( - `INSERT INTO "${ref.schema}"."${ref.table}" - (id, database_id, entity_id, actor_id, request_id, - model, provider, service, operation, - input_tokens, output_tokens, total_tokens, - latency_ms, status, error_type, raw_usage, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`, - [ - id, - entry.databaseId ?? null, - entry.entityId ?? null, - entry.actorId ?? null, - entry.requestId ?? randomUUID(), - entry.model, - entry.provider, - entry.service, - entry.operation, - entry.inputTokens, - entry.outputTokens, - entry.totalTokens, - Math.round(entry.latencyMs), - entry.status, - entry.errorType ?? null, - entry.rawUsage ? JSON.stringify(entry.rawUsage) : null, - now, - ] - ) - .catch((err) => { - log.warn(`inference log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); - }) - .catch((err) => { - log.warn(`inference usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); - }); +/** Reset the module-level cache (for testing). */ +export function _resetCache(): void { + _loader = null; + _pool = null; } diff --git a/packages/agentic-server/src/router.ts b/packages/agentic-server/src/router.ts index 08d76445c..c24d495a8 100644 --- a/packages/agentic-server/src/router.ts +++ b/packages/agentic-server/src/router.ts @@ -5,137 +5,297 @@ import { logInferenceUsage } from './inference-meter'; const log = new Logger('agentic-server'); +// ─── Provider Configuration ─────────────────────────────────────────────────── + +export interface ProviderConfig { + /** Provider name (e.g. 'openai', 'anthropic', 'ollama') */ + type: string; + /** Base URL for the provider API */ + baseUrl: string; + /** API key for authentication (optional for local providers like Ollama) */ + apiKey?: string; + /** Default model for this provider */ + defaultModel?: string; +} + export interface AgenticRouterOptions { - /** Upstream LLM provider base URL (e.g., https://api.openai.com, http://ollama:11434) */ - providerBaseUrl: string; - /** API key for the upstream provider (if needed) */ + /** + * Primary provider configuration. + * Legacy: if providerBaseUrl is set directly, it takes precedence. + */ + providers?: ProviderConfig[]; + /** Legacy: single upstream LLM provider base URL */ + providerBaseUrl?: string; + /** Legacy: API key for the upstream provider */ providerApiKey?: string; - /** Default model name */ + /** Legacy: Default model name */ defaultModel?: string; - /** Provider type: 'openai' | 'ollama' | 'anthropic' */ + /** Legacy: Provider type: 'openai' | 'ollama' | 'anthropic' */ providerType?: string; /** Optional pg pool for inference metering (fire-and-forget writes) */ pgPool?: Pool; } +// ─── Provider Resolution ────────────────────────────────────────────────────── + +interface ResolvedProvider { + type: string; + baseUrl: string; + apiKey?: string; + defaultModel?: string; +} + /** - * Create an Express router that proxies OpenAI-compatible requests - * to a configured LLM provider. - * - * Identity headers (X-Database-Id, X-Entity-Id, X-Actor-Id) are trusted - * ONLY when X-Internal-Service is present (i.e., the request comes from - * fn-runtime, not from an external client). External requests must - * authenticate via other means. + * Resolve which provider to use for a given request. + * Priority: model-based routing → explicit provider header → default. */ -export const createRouter = (options: AgenticRouterOptions): Router => { - const router = Router(); - const { providerBaseUrl, providerApiKey, defaultModel, providerType, pgPool } = options; - - // Resolve upstream URL based on provider type - const resolveUpstreamUrl = (path: string): string => { - const type = providerType || 'openai'; - if (type === 'ollama') { - if (path === '/v1/chat/completions') return `${providerBaseUrl}/api/chat`; - if (path === '/v1/embeddings') return `${providerBaseUrl}/api/embed`; +function resolveProvider( + options: AgenticRouterOptions, + requestModel?: string, + requestProvider?: string +): ResolvedProvider { + const providers = options.providers || []; + + // If multi-provider is configured, route by provider name or model prefix + if (providers.length > 0) { + // Check X-Provider header first + if (requestProvider) { + const match = providers.find((p) => p.type === requestProvider); + if (match) return match; + } + + // Route by model prefix: "anthropic/claude-3" → anthropic provider + if (requestModel && requestModel.includes('/')) { + const prefix = requestModel.split('/')[0]; + const match = providers.find((p) => p.type === prefix); + if (match) { + return { ...match, defaultModel: requestModel.split('/').slice(1).join('/') }; + } } - // OpenAI-compatible (OpenAI, Anthropic via proxy, etc.) - return `${providerBaseUrl}${path}`; - }; - // Transform request body for different provider types - const transformRequest = (body: Record, path: string): Record => { - const type = providerType || 'openai'; - if (type === 'ollama') { - if (path === '/v1/chat/completions') { - return { - model: body.model || defaultModel || 'llama3', - messages: body.messages, - stream: false, - ...(body.temperature !== undefined && { - options: { temperature: body.temperature } - }) - }; + // Route by known model patterns + if (requestModel) { + if (requestModel.startsWith('claude')) { + const match = providers.find((p) => p.type === 'anthropic'); + if (match) return match; + } + if (requestModel.startsWith('gpt') || requestModel.startsWith('o1') || requestModel.startsWith('o3')) { + const match = providers.find((p) => p.type === 'openai'); + if (match) return match; } - if (path === '/v1/embeddings') { - return { - model: body.model || defaultModel || 'nomic-embed-text', - input: body.input - }; + if (requestModel.startsWith('llama') || requestModel.startsWith('mistral') || requestModel.startsWith('gemma')) { + const match = providers.find((p) => p.type === 'ollama'); + if (match) return match; } } - // OpenAI format — pass through with default model + + // Fall back to first provider + return providers[0]; + } + + // Legacy single-provider mode + return { + type: options.providerType || 'openai', + baseUrl: options.providerBaseUrl || 'http://localhost:11434', + apiKey: options.providerApiKey, + defaultModel: options.defaultModel + }; +} + +// ─── Provider-Specific Transforms ───────────────────────────────────────────── + +function resolveUpstreamUrl(provider: ResolvedProvider, path: string): string { + if (provider.type === 'ollama') { + if (path === '/v1/chat/completions') return `${provider.baseUrl}/api/chat`; + if (path === '/v1/embeddings') return `${provider.baseUrl}/api/embed`; + } + if (provider.type === 'anthropic') { + if (path === '/v1/chat/completions') return `${provider.baseUrl}/v1/messages`; + } + return `${provider.baseUrl}${path}`; +} + +function buildHeaders(provider: ResolvedProvider): Record { + const headers: Record = { 'Content-Type': 'application/json' }; + if (provider.apiKey) { + if (provider.type === 'anthropic') { + headers['x-api-key'] = provider.apiKey; + headers['anthropic-version'] = '2023-06-01'; + } else { + headers['Authorization'] = `Bearer ${provider.apiKey}`; + } + } + return headers; +} + +function transformChatRequest( + provider: ResolvedProvider, + body: Record +): Record { + // Strip provider prefix from model name: "anthropic/claude-3" → "claude-3" + let model = body.model || provider.defaultModel; + if (typeof model === 'string' && model.includes('/')) { + const prefix = model.split('/')[0]; + if (prefix === provider.type) { + model = model.split('/').slice(1).join('/'); + } + } + + if (provider.type === 'ollama') { return { - ...body, - model: body.model || defaultModel || 'gpt-4o' + model: model || 'llama3', + messages: body.messages, + stream: false, + ...(body.temperature !== undefined && { + options: { temperature: body.temperature } + }) }; - }; + } + + if (provider.type === 'anthropic') { + const messages = body.messages as Array<{ role: string; content: string }> | undefined; + const systemMsg = messages?.find((m) => m.role === 'system'); + const nonSystem = messages?.filter((m) => m.role !== 'system') || []; + return { + model: model || 'claude-sonnet-4-20250514', + messages: nonSystem, + max_tokens: body.max_tokens || 4096, + ...(systemMsg && { system: systemMsg.content }), + ...(body.temperature !== undefined && { temperature: body.temperature }) + }; + } + + // OpenAI-compatible (default) + return { ...body, model: model || 'gpt-4o' }; +} + +function transformEmbedRequest( + provider: ResolvedProvider, + body: Record +): Record { + const model = body.model || provider.defaultModel; + + if (provider.type === 'ollama') { + return { + model: model || 'nomic-embed-text', + input: body.input + }; + } + return { ...body, model: model || 'text-embedding-3-small' }; +} - // Transform provider response to OpenAI-compatible format - const transformChatResponse = (data: Record, type: string): Record => { - if (type === 'ollama') { - const msg = data.message as { role: string; content: string } | undefined; - return { +interface UsageResult { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; +} + +function transformChatResponse( + data: Record, + provider: ResolvedProvider +): { body: Record; usage: UsageResult } { + if (provider.type === 'ollama') { + const msg = data.message as { role: string; content: string } | undefined; + const promptTokens = (data.prompt_eval_count as number) || 0; + const completionTokens = (data.eval_count as number) || 0; + const usage = { prompt_tokens: promptTokens, completion_tokens: completionTokens, total_tokens: promptTokens + completionTokens }; + return { + body: { id: `chatcmpl-${Date.now()}`, object: 'chat.completion', - choices: [ - { - message: msg || { role: 'assistant', content: '' }, - finish_reason: 'stop', - index: 0 - } - ], - usage: { - prompt_tokens: (data.prompt_eval_count as number) || 0, - completion_tokens: (data.eval_count as number) || 0, - total_tokens: ((data.prompt_eval_count as number) || 0) + ((data.eval_count as number) || 0) - } - }; - } - return data; - }; + choices: [{ message: msg || { role: 'assistant', content: '' }, finish_reason: 'stop', index: 0 }], + usage + }, + usage + }; + } + + if (provider.type === 'anthropic') { + const content = data.content as Array<{ type: string; text?: string }> | undefined; + const text = content?.find((c) => c.type === 'text')?.text || ''; + const inputUsage = (data.usage as Record) || {}; + const usage = { + prompt_tokens: inputUsage.input_tokens || 0, + completion_tokens: inputUsage.output_tokens || 0, + total_tokens: (inputUsage.input_tokens || 0) + (inputUsage.output_tokens || 0) + }; + return { + body: { + id: data.id || `chatcmpl-${Date.now()}`, + object: 'chat.completion', + choices: [{ message: { role: 'assistant', content: text }, finish_reason: 'stop', index: 0 }], + usage + }, + usage + }; + } - const transformEmbedResponse = (data: Record, type: string): Record => { - if (type === 'ollama') { - const raw = (data.embeddings || data.embedding || []) as number[][]; - const embeddings: number[][] = Array.isArray(raw[0]) ? raw : [raw as unknown as number[]]; - return { + // OpenAI — pass through + const usage = (data.usage || {}) as UsageResult; + return { body: data, usage }; +} + +function transformEmbedResponse( + data: Record, + provider: ResolvedProvider +): { body: Record; usage: UsageResult } { + if (provider.type === 'ollama') { + const raw = (data.embeddings || data.embedding || []) as number[][]; + const embeddings: number[][] = Array.isArray(raw[0]) ? raw : [raw as unknown as number[]]; + const usage = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }; + return { + body: { object: 'list', - data: embeddings.map((emb, i) => ({ - object: 'embedding', - embedding: emb, - index: i - })), - usage: { prompt_tokens: 0, total_tokens: 0 } - }; - } - return data; - }; + data: embeddings.map((emb, i) => ({ object: 'embedding', embedding: emb, index: i })), + usage + }, + usage + }; + } + + const usage = (data.usage || {}) as UsageResult; + return { body: data, usage }; +} + +// ─── Router ─────────────────────────────────────────────────────────────────── + +/** + * Create an Express router that proxies OpenAI-compatible requests + * to configured LLM provider(s) with multi-provider routing. + * + * Supports: + * - Multiple providers (OpenAI, Anthropic, Ollama) configured at startup + * - Model-based routing: "anthropic/claude-3.5-sonnet" → anthropic provider + * - Header-based routing: X-Provider: ollama → ollama provider + * - Known model prefix routing: claude-* → anthropic, gpt-* → openai + * - Fire-and-forget inference metering on all calls + * - /v1/usage reporting endpoint for external usage submission + */ +export const createRouter = (options: AgenticRouterOptions): Router => { + const router = Router(); + const { pgPool } = options; - // POST /v1/chat/completions — proxy to LLM provider + // POST /v1/chat/completions — multi-provider LLM proxy router.post('/v1/chat/completions', async (req: any, res: any) => { - const internalService = req.get('X-Internal-Service'); const databaseId = req.get('X-Database-Id'); const entityId = req.get('X-Entity-Id'); const actorId = req.get('X-Actor-Id'); + const requestProvider = req.get('X-Provider'); const startTime = process.hrtime.bigint(); + const provider = resolveProvider(options, req.body?.model, requestProvider); + log.info('chat/completions', { - internal: !!internalService, databaseId, entityId, + provider: provider.type, model: req.body?.model }); try { - const upstreamUrl = resolveUpstreamUrl('/v1/chat/completions'); - const body = transformRequest(req.body || {}, '/v1/chat/completions'); - - const headers: Record = { - 'Content-Type': 'application/json' - }; - if (providerApiKey) { - headers['Authorization'] = `Bearer ${providerApiKey}`; - } + const upstreamUrl = resolveUpstreamUrl(provider, '/v1/chat/completions'); + const body = transformChatRequest(provider, req.body || {}); + const headers = buildHeaders(provider); const upstream = await fetch(upstreamUrl, { method: 'POST', @@ -147,13 +307,13 @@ export const createRouter = (options: AgenticRouterOptions): Router => { if (!upstream.ok) { const text = await upstream.text().catch(() => ''); - log.error('upstream error', { status: upstream.status, body: text }); + log.error('upstream error', { status: upstream.status, provider: provider.type, body: text }); if (pgPool) { logInferenceUsage(pgPool, { databaseId, entityId, actorId, - model: String(body.model || ''), - provider: providerType || 'openai', + model: String(req.body?.model || body.model || ''), + provider: provider.type, service: 'chat', operation: 'chat/completions', inputTokens: 0, outputTokens: 0, totalTokens: 0, @@ -169,25 +329,22 @@ export const createRouter = (options: AgenticRouterOptions): Router => { return; } - const type = providerType || 'openai'; const data = await upstream.json() as Record; - const response = transformChatResponse(data, type); + const { body: responseBody, usage } = transformChatResponse(data, provider); - const usage = (response.usage || {}) as Record; log.info('inference complete', { databaseId, - entityId, - model: body.model, + provider: provider.type, + model: req.body?.model, promptTokens: usage.prompt_tokens, - completionTokens: usage.completion_tokens, - totalTokens: usage.total_tokens + completionTokens: usage.completion_tokens }); if (pgPool) { logInferenceUsage(pgPool, { databaseId, entityId, actorId, - model: String(body.model || ''), - provider: providerType || 'openai', + model: String(req.body?.model || body.model || ''), + provider: provider.type, service: 'chat', operation: 'chat/completions', inputTokens: usage.prompt_tokens || 0, @@ -199,7 +356,7 @@ export const createRouter = (options: AgenticRouterOptions): Router => { }); } - res.json(response); + res.json(responseBody); } catch (err: any) { const latencyMs = Number(process.hrtime.bigint() - startTime) / 1e6; log.error('chat/completions error', err); @@ -208,7 +365,7 @@ export const createRouter = (options: AgenticRouterOptions): Router => { logInferenceUsage(pgPool, { databaseId, entityId, actorId, model: String(req.body?.model || ''), - provider: providerType || 'openai', + provider: provider.type, service: 'chat', operation: 'chat/completions', inputTokens: 0, outputTokens: 0, totalTokens: 0, @@ -224,25 +381,22 @@ export const createRouter = (options: AgenticRouterOptions): Router => { } }); - // POST /v1/embeddings — proxy to LLM provider + // POST /v1/embeddings — multi-provider embedding proxy router.post('/v1/embeddings', async (req: any, res: any) => { const databaseId = req.get('X-Database-Id'); const entityId = req.get('X-Entity-Id'); const actorId = req.get('X-Actor-Id'); + const requestProvider = req.get('X-Provider'); const startTime = process.hrtime.bigint(); - log.info('embeddings', { databaseId, entityId, model: req.body?.model }); + const provider = resolveProvider(options, req.body?.model, requestProvider); + + log.info('embeddings', { databaseId, entityId, provider: provider.type, model: req.body?.model }); try { - const upstreamUrl = resolveUpstreamUrl('/v1/embeddings'); - const body = transformRequest(req.body || {}, '/v1/embeddings'); - - const headers: Record = { - 'Content-Type': 'application/json' - }; - if (providerApiKey) { - headers['Authorization'] = `Bearer ${providerApiKey}`; - } + const upstreamUrl = resolveUpstreamUrl(provider, '/v1/embeddings'); + const body = transformEmbedRequest(provider, req.body || {}); + const headers = buildHeaders(provider); const upstream = await fetch(upstreamUrl, { method: 'POST', @@ -254,13 +408,13 @@ export const createRouter = (options: AgenticRouterOptions): Router => { if (!upstream.ok) { const text = await upstream.text().catch(() => ''); - log.error('upstream embed error', { status: upstream.status, body: text }); + log.error('upstream embed error', { status: upstream.status, provider: provider.type }); if (pgPool) { logInferenceUsage(pgPool, { databaseId, entityId, actorId, - model: String(body.model || ''), - provider: providerType || 'openai', + model: String(req.body?.model || body.model || ''), + provider: provider.type, service: 'embed', operation: 'embeddings', inputTokens: 0, outputTokens: 0, totalTokens: 0, @@ -276,18 +430,16 @@ export const createRouter = (options: AgenticRouterOptions): Router => { return; } - const type = providerType || 'openai'; const data = await upstream.json() as Record; - const response = transformEmbedResponse(data, type); + const { body: responseBody, usage } = transformEmbedResponse(data, provider); - const usage = (response.usage || {}) as Record; - log.info('embed complete', { databaseId, entityId }); + log.info('embed complete', { databaseId, entityId, provider: provider.type }); if (pgPool) { logInferenceUsage(pgPool, { databaseId, entityId, actorId, - model: String(body.model || ''), - provider: providerType || 'openai', + model: String(req.body?.model || body.model || ''), + provider: provider.type, service: 'embed', operation: 'embeddings', inputTokens: usage.prompt_tokens || 0, @@ -299,7 +451,7 @@ export const createRouter = (options: AgenticRouterOptions): Router => { }); } - res.json(response); + res.json(responseBody); } catch (err: any) { const latencyMs = Number(process.hrtime.bigint() - startTime) / 1e6; log.error('embeddings error', err); @@ -308,7 +460,7 @@ export const createRouter = (options: AgenticRouterOptions): Router => { logInferenceUsage(pgPool, { databaseId, entityId, actorId, model: String(req.body?.model || ''), - provider: providerType || 'openai', + provider: provider.type, service: 'embed', operation: 'embeddings', inputTokens: 0, outputTokens: 0, totalTokens: 0, @@ -324,12 +476,80 @@ export const createRouter = (options: AgenticRouterOptions): Router => { } }); + // POST /v1/usage — external usage reporting endpoint + // Allows Python functions (LlamaParse, HuggingFace, etc.) to report inference + // usage back for billing when they call local models directly. + router.post('/v1/usage', (req: any, res: any) => { + const databaseId = req.get('X-Database-Id'); + const entityId = req.get('X-Entity-Id'); + const actorId = req.get('X-Actor-Id'); + + const { + model, + provider: reportedProvider, + service, + operation, + input_tokens, + output_tokens, + total_tokens, + latency_ms, + status, + error_type, + raw_usage + } = req.body || {}; + + if (!model) { + res.status(400).json({ error: { message: 'model is required' } }); + return; + } + + if (pgPool) { + logInferenceUsage(pgPool, { + databaseId, + entityId, + actorId, + model: String(model), + provider: String(reportedProvider || 'unknown'), + service: (service === 'embed' ? 'embed' : 'chat') as 'chat' | 'embed', + operation: operation || 'external', + inputTokens: Number(input_tokens) || 0, + outputTokens: Number(output_tokens) || 0, + totalTokens: Number(total_tokens) || 0, + latencyMs: Number(latency_ms) || 0, + status: status === 'error' ? 'error' : 'ok', + errorType: error_type || undefined, + rawUsage: raw_usage || undefined + }); + } + + res.status(202).json({ accepted: true }); + }); + + // GET /v1/providers — list configured providers + router.get('/v1/providers', (_req: any, res: any) => { + const providers = options.providers || [{ + type: options.providerType || 'openai', + baseUrl: options.providerBaseUrl || 'http://localhost:11434' + }]; + res.json({ + providers: providers.map((p) => ({ + type: p.type, + defaultModel: p.defaultModel + })) + }); + }); + // GET /healthz — health check router.get('/healthz', (_req: any, res: any) => { + const providers = options.providers || [{ + type: options.providerType || 'openai', + baseUrl: options.providerBaseUrl || 'http://localhost:11434' + }]; res.json({ status: 'ok', - provider: providerType || 'openai', - providerUrl: providerBaseUrl + provider: providers[0]?.type || 'openai', + providerUrl: providers[0]?.baseUrl || options.providerBaseUrl, + providers: providers.map((p) => p.type) }); }); diff --git a/packages/agentic-server/src/standalone.ts b/packages/agentic-server/src/standalone.ts index 6d0f8fd66..bc1c7c5fb 100644 --- a/packages/agentic-server/src/standalone.ts +++ b/packages/agentic-server/src/standalone.ts @@ -1,34 +1,55 @@ /** * Standalone entry point for the agentic server. * - * Reads configuration from environment variables: + * Configuration via environment variables: * PORT — Server port (default: 3003) * LLM_PROVIDER_URL — Upstream LLM provider base URL (required) * LLM_PROVIDER_API_KEY — API key for the upstream provider * LLM_PROVIDER_TYPE — Provider type: openai | ollama | anthropic (default: openai) * LLM_DEFAULT_MODEL — Default model name (default: gpt-4o) + * + * Multi-provider (optional — overrides single provider): + * LLM_PROVIDERS — JSON array of provider configs, e.g.: + * [{"type":"openai","baseUrl":"https://api.openai.com/v1","apiKey":"sk-..."}, + * {"type":"anthropic","baseUrl":"https://api.anthropic.com","apiKey":"sk-ant-..."}, + * {"type":"ollama","baseUrl":"http://localhost:11434"}] */ import { createAgenticServer } from './server'; import { Logger } from '@pgpmjs/logger'; +import type { ProviderConfig } from './router'; const log = new Logger('agentic-server'); const port = Number(process.env.PORT || 3003); + +// Multi-provider config (takes precedence) +let providers: ProviderConfig[] | undefined; +if (process.env.LLM_PROVIDERS) { + try { + providers = JSON.parse(process.env.LLM_PROVIDERS); + log.info(`Multi-provider mode: ${providers!.map((p) => p.type).join(', ')}`); + } catch (e) { + log.error('Failed to parse LLM_PROVIDERS — falling back to single provider'); + } +} + +// Single provider fallback const providerBaseUrl = process.env.LLM_PROVIDER_URL || 'http://localhost:11434'; const providerApiKey = process.env.LLM_PROVIDER_API_KEY; const providerType = process.env.LLM_PROVIDER_TYPE || 'ollama'; const defaultModel = process.env.LLM_DEFAULT_MODEL; const app = createAgenticServer({ - providerBaseUrl, - providerApiKey, - providerType, - defaultModel + ...(providers ? { providers } : { providerBaseUrl, providerApiKey, providerType, defaultModel }) }); app.listen(port, () => { log.info(`Agentic server listening on port ${port}`); - log.info(`Provider: ${providerType} @ ${providerBaseUrl}`); + if (providers) { + providers.forEach((p) => log.info(` ${p.type} @ ${p.baseUrl}`)); + } else { + log.info(`Provider: ${providerType} @ ${providerBaseUrl}`); + } if (defaultModel) log.info(`Default model: ${defaultModel}`); }); diff --git a/packages/fn-runtime/package.json b/packages/fn-runtime/package.json index 076ff67e7..8b7abd89d 100644 --- a/packages/fn-runtime/package.json +++ b/packages/fn-runtime/package.json @@ -21,6 +21,7 @@ "@aws-sdk/client-s3": "^3.1066.0", "@constructive-io/fn-types": "workspace:^", "@constructive-io/knative-job-fn": "workspace:^", + "@constructive-io/usage-loader": "workspace:^", "@pgpmjs/logger": "^2.4.3", "graphql-request": "^7.1.2" }, diff --git a/packages/fn-runtime/src/storage.ts b/packages/fn-runtime/src/storage.ts index 8404a6805..2514cf1bc 100644 --- a/packages/fn-runtime/src/storage.ts +++ b/packages/fn-runtime/src/storage.ts @@ -5,10 +5,7 @@ import { PutObjectCommand, DeleteObjectCommand } from '@aws-sdk/client-s3'; -import { Logger } from '@pgpmjs/logger'; -import { randomUUID } from 'crypto'; - -const meterLog = new Logger('storage-meter'); +import { UsageLoader } from '@constructive-io/usage-loader'; export type StorageMeterCallback = (info: { databaseId?: string; @@ -22,9 +19,10 @@ export type StorageMeterCallback = (info: { }) => void; /** - * Create a fire-and-forget storage metering callback backed by pg. + * Create a fire-and-forget storage metering callback backed by UsageLoader. * * Lazily creates a pg Pool from standard PG* env vars on first invocation. + * Resolves table names dynamically from MetaSchema (scope-aware). * Returns undefined if PGHOST/DATABASE_URL is not set (metering disabled). */ export const createMeterCallback = (): StorageMeterCallback | undefined => { @@ -32,38 +30,27 @@ export const createMeterCallback = (): StorageMeterCallback | undefined => { if (!env.PGHOST && !env.DATABASE_URL) return undefined; let pool: import('pg').Pool | undefined; + let loader: UsageLoader | undefined; - const getPool = (): import('pg').Pool => { - if (pool) return pool; + const getLoader = (): UsageLoader => { + if (loader) return loader; // eslint-disable-next-line @typescript-eslint/no-require-imports const { Pool } = require('pg') as typeof import('pg'); pool = new Pool({ max: 2 }); - return pool; + loader = new UsageLoader(pool); + return loader; }; return (info) => { - const p = getPool(); - const id = randomUUID(); - const now = new Date(); - p.query( - `INSERT INTO "constructive_usage_public".platform_usage_log_storage - (id, database_id, entity_id, actor_id, operation, - bucket, key, size_bytes, duration_ms, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, - [ - id, - info.databaseId ?? null, - info.entityId ?? null, - info.actorId ?? null, - info.operation, - info.bucket, - info.key, - info.sizeBytes, - Math.round(info.durationMs), - now - ] - ).catch((err: unknown) => { - meterLog.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + getLoader().logStorageUsage({ + databaseId: info.databaseId, + entityId: info.entityId, + actorId: info.actorId, + operation: info.operation, + bucket: info.bucket, + key: info.key, + sizeBytes: info.sizeBytes, + durationMs: info.durationMs }); }; }; diff --git a/packages/module-loader/package.json b/packages/module-loader/package.json new file mode 100644 index 000000000..b7ed8ad18 --- /dev/null +++ b/packages/module-loader/package.json @@ -0,0 +1,26 @@ +{ + "name": "@constructive-io/module-loader", + "version": "1.0.0", + "description": "Unified MetaSchema module loader — resolves compute, usage, and billing table/schema names with TTL caching. Multi-database (platform + tenant) aware.", + "author": "Constructive ", + "license": "SEE LICENSE IN LICENSE", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "files": [ + "dist", + "README.md" + ], + "publishConfig": { + "access": "public" + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "clean": "rimraf dist" + }, + "dependencies": { + "@pgpmjs/logger": "^2.4.3" + }, + "peerDependencies": { + "pg": "^8.0.0" + } +} diff --git a/packages/module-loader/src/billing-loader.ts b/packages/module-loader/src/billing-loader.ts new file mode 100644 index 000000000..2d3b0a0e1 --- /dev/null +++ b/packages/module-loader/src/billing-loader.ts @@ -0,0 +1,127 @@ +/** + * BillingLoader — quota checks and usage recording via the billing_module. + * + * Discovers billing configuration from metaschema_modules_public.billing_module. + * Gracefully no-ops when billing is not provisioned (standalone dev mode). + * + * Multi-database: pass different databaseId to resolve billing for + * platform scope vs tenant scope. + */ + +import { Logger } from '@pgpmjs/logger'; +import type { Pool } from 'pg'; + +import { TtlCache } from './cache'; +import type { BillingModuleConfig } from './types'; +import { DEFAULT_DATABASE_ID, DEFAULT_TTL_MS } from './types'; + +const log = new Logger('module-loader:billing'); + +const BILLING_MODULE_SQL = ` + SELECT + s.schema_name AS public_schema, + ps.schema_name AS private_schema, + bm.record_usage_function + FROM metaschema_modules_public.billing_module bm + JOIN metaschema_public.schema s ON bm.schema_id = s.id + JOIN metaschema_public.schema ps ON bm.private_schema_id = ps.id + WHERE bm.database_id = $1 + LIMIT 1 +`; + +export class BillingLoader { + private pool: Pool; + private cache: TtlCache; + private defaultDatabaseId: string; + + constructor(pool: Pool, databaseId: string = DEFAULT_DATABASE_ID, ttlMs: number = DEFAULT_TTL_MS) { + this.pool = pool; + this.defaultDatabaseId = databaseId; + this.cache = new TtlCache(ttlMs); + } + + /** + * Load billing module config for a database. + * Returns null if billing is not provisioned. + */ + async load(databaseId?: string): Promise { + const dbId = databaseId ?? this.defaultDatabaseId; + const cached = this.cache.get(dbId); + if (cached !== undefined) return cached; + + try { + const { rows } = await this.pool.query(BILLING_MODULE_SQL, [dbId]); + if (!rows.length || !rows[0].record_usage_function) { + this.cache.set(dbId, null); + return null; + } + const config: BillingModuleConfig = { + publicSchema: rows[0].public_schema, + privateSchema: rows[0].private_schema, + recordUsageFunction: rows[0].record_usage_function, + }; + log.debug(`loaded billing module: ${config.privateSchema}.${config.recordUsageFunction}`); + this.cache.set(dbId, config); + return config; + } catch { + this.cache.set(dbId, null); + return null; + } + } + + /** + * Check if the entity has quota for this meter. + * Returns true if billing is not provisioned (graceful degradation). + */ + async checkQuota( + entityId: string, + meterSlug: string, + amount: number = 1, + databaseId?: string + ): Promise { + const config = await this.load(databaseId); + if (!config) return true; + + try { + const sql = `SELECT "${config.privateSchema}"."check_billing_quota"($1, $2::uuid, $3) AS allowed`; + const { rows } = await this.pool.query(sql, [meterSlug, entityId, amount]); + return rows[0]?.allowed !== false; + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : String(err); + log.warn(`check_billing_quota failed (allowing): ${msg}`); + return true; + } + } + + /** + * Record usage after a function completes. + * No-ops if billing is not provisioned. + */ + async recordUsage( + entityId: string, + meterSlug: string, + amount: number, + metadata: Record, + databaseId?: string + ): Promise { + const config = await this.load(databaseId); + if (!config) return; + + try { + const sql = `SELECT "${config.privateSchema}"."${config.recordUsageFunction}"($1, $2::uuid, $3, $4::jsonb)`; + await this.pool.query(sql, [meterSlug, entityId, amount, JSON.stringify(metadata)]); + log.debug(`recorded usage: ${meterSlug} entity=${entityId} amount=${amount}`); + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : String(err); + log.warn(`record_usage failed (non-fatal): ${msg}`); + } + } + + invalidate(databaseId?: string): void { + if (databaseId) { + this.cache.delete(databaseId); + } else { + this.cache.clear(); + } + } +} diff --git a/packages/module-loader/src/cache.ts b/packages/module-loader/src/cache.ts new file mode 100644 index 000000000..d9001c630 --- /dev/null +++ b/packages/module-loader/src/cache.ts @@ -0,0 +1,47 @@ +/** + * TtlCache — generic time-to-live cache. + * + * Used by all module loaders to cache MetaSchema-resolved configs. + * Entries expire after the configured TTL. Thread-safe for single-threaded + * Node.js (no concurrent writes to same key). + */ + +interface CacheEntry { + value: T; + expiresAt: number; +} + +export class TtlCache { + private store = new Map>(); + private ttlMs: number; + + constructor(ttlMs: number) { + this.ttlMs = ttlMs; + } + + get(key: string): T | undefined { + const entry = this.store.get(key); + if (!entry) return undefined; + if (Date.now() > entry.expiresAt) { + this.store.delete(key); + return undefined; + } + return entry.value; + } + + set(key: string, value: T): void { + this.store.set(key, { value, expiresAt: Date.now() + this.ttlMs }); + } + + delete(key: string): void { + this.store.delete(key); + } + + clear(): void { + this.store.clear(); + } + + get size(): number { + return this.store.size; + } +} diff --git a/packages/module-loader/src/compute-loader.ts b/packages/module-loader/src/compute-loader.ts new file mode 100644 index 000000000..14fc2c7c3 --- /dev/null +++ b/packages/module-loader/src/compute-loader.ts @@ -0,0 +1,209 @@ +/** + * ComputeModuleLoader — resolves compute module schema/table names + * dynamically from MetaSchema. + * + * Queries metaschema_modules_public.function_module, + * metaschema_modules_public.function_invocation_module, and + * metaschema_modules_public.compute_log_module, joined with + * metaschema_public.schema to resolve schema names. + * + * Results are TTL-cached per database_id for multi-tenant support. + */ + +import { Logger } from '@pgpmjs/logger'; +import type { Pool } from 'pg'; + +import { TtlCache } from './cache'; +import type { + ComputeLogModuleConfig, + ComputeModuleConfig, + FunctionModuleConfig, + GraphExecutionModuleConfig, + InvocationModuleConfig, +} from './types'; +import { DEFAULT_TTL_MS } from './types'; + +const log = new Logger('module-loader:compute'); + +// ─── SQL Queries ───────────────────────────────────────────────────────────── + +const FUNCTION_MODULE_SQL = ` + SELECT + s.schema_name AS public_schema, + ps.schema_name AS private_schema, + fm.definitions_table_name, + fm.secret_definitions_table_name, + fm.scope + FROM metaschema_modules_public.function_module fm + JOIN metaschema_public.schema s ON fm.schema_id = s.id + JOIN metaschema_public.schema ps ON fm.private_schema_id = ps.id + WHERE fm.database_id = $1 +`; + +const INVOCATION_MODULE_SQL = ` + SELECT + s.schema_name AS public_schema, + fim.invocations_table_name, + fim.execution_logs_table_name, + fim.scope + FROM metaschema_modules_public.function_invocation_module fim + JOIN metaschema_public.schema s ON fim.schema_id = s.id + WHERE fim.database_id = $1 +`; + +const COMPUTE_LOG_MODULE_SQL = ` + SELECT + s.schema_name AS public_schema, + ps.schema_name AS private_schema, + clm.compute_log_table_name, + clm.usage_daily_table_name, + clm.scope + FROM metaschema_modules_public.compute_log_module clm + JOIN metaschema_public.schema s ON clm.schema_id = s.id + JOIN metaschema_public.schema ps ON clm.private_schema_id = ps.id + WHERE clm.database_id = $1 +`; + +const GRAPH_EXECUTION_MODULE_SQL = ` + SELECT + s.schema_name AS public_schema, + ps.schema_name AS private_schema, + gem.node_states_table_name, + gem.complete_node_function, + gem.fail_node_function + FROM metaschema_modules_public.graph_execution_module gem + JOIN metaschema_public.schema s ON gem.schema_id = s.id + JOIN metaschema_public.schema ps ON gem.private_schema_id = ps.id + WHERE gem.database_id = $1 +`; + +// ─── Default Configs ───────────────────────────────────────────────────────── + +const DEFAULT_GRAPH_EXECUTION: GraphExecutionModuleConfig = { + publicSchema: 'constructive_compute_public', + privateSchema: 'constructive_compute_private', + nodeStatesTable: 'platform_function_graph_execution_node_states', + completeNodeFunction: 'platform_complete_node', + failNodeFunction: 'platform_fail_node', +}; + +// ─── Loader Class ──────────────────────────────────────────────────────────── + +export interface ComputeModuleConfigExtended extends ComputeModuleConfig { + graphExecutionModule: GraphExecutionModuleConfig; +} + +export class ComputeModuleLoader { + private cache: TtlCache; + private pool: Pool; + + constructor(pool: Pool, ttlMs: number = DEFAULT_TTL_MS) { + this.pool = pool; + this.cache = new TtlCache(ttlMs); + } + + /** + * Load the full compute module config for a database. + * Cached per database_id. Safe for multi-tenant usage: + * - `load(platformId)` → platform scope + * - `load(tenantId)` → tenant scope + */ + async load(databaseId: string): Promise { + const cached = this.cache.get(databaseId); + if (cached !== undefined) { + log.debug(`module config cache hit for database ${databaseId}`); + return cached; + } + + log.debug(`module config cache miss for database ${databaseId}, querying metaschema`); + + const fnResult = await this.pool.query(FUNCTION_MODULE_SQL, [databaseId]); + + let functionModule: FunctionModuleConfig | null = null; + if (fnResult.rows.length > 0) { + const row = fnResult.rows[0]; + functionModule = { + publicSchema: row.public_schema, + privateSchema: row.private_schema, + definitionsTable: row.definitions_table_name, + secretDefinitionsTable: row.secret_definitions_table_name, + scope: row.scope, + }; + } + + let invocationModules: InvocationModuleConfig[] = []; + try { + const invResult = await this.pool.query(INVOCATION_MODULE_SQL, [databaseId]); + invocationModules = invResult.rows.map( + (row: Record) => ({ + publicSchema: row.public_schema, + invocationsTable: row.invocations_table_name, + executionLogsTable: row.execution_logs_table_name, + scope: row.scope, + }) + ); + } catch { + log.debug(`function_invocation_module not available for database ${databaseId}`); + } + + let computeLogModule: ComputeLogModuleConfig | null = null; + try { + const clResult = await this.pool.query(COMPUTE_LOG_MODULE_SQL, [databaseId]); + if (clResult.rows.length > 0) { + const row = clResult.rows[0]; + computeLogModule = { + publicSchema: row.public_schema, + privateSchema: row.private_schema, + computeLogTable: row.compute_log_table_name, + usageDailyTable: row.usage_daily_table_name, + scope: row.scope, + }; + } + } catch { + log.debug(`compute_log_module not available for database ${databaseId}`); + } + + let graphExecutionModule: GraphExecutionModuleConfig = DEFAULT_GRAPH_EXECUTION; + try { + const geResult = await this.pool.query(GRAPH_EXECUTION_MODULE_SQL, [databaseId]); + if (geResult.rows.length > 0) { + const row = geResult.rows[0]; + graphExecutionModule = { + publicSchema: row.public_schema, + privateSchema: row.private_schema, + nodeStatesTable: row.node_states_table_name, + completeNodeFunction: row.complete_node_function, + failNodeFunction: row.fail_node_function, + }; + } + } catch { + log.debug(`graph_execution_module not available for database ${databaseId} — using defaults`); + } + + const config: ComputeModuleConfigExtended = { + functionModule, + invocationModules, + computeLogModule, + graphExecutionModule, + }; + this.cache.set(databaseId, config); + + log.info( + `loaded compute module config for database ${databaseId}: ` + + `fn=${functionModule ? `${functionModule.publicSchema}.${functionModule.definitionsTable}` : 'none'}, ` + + `invocations=${invocationModules.length} scope(s), ` + + `computeLog=${computeLogModule ? `${computeLogModule.publicSchema}.${computeLogModule.computeLogTable}` : 'none'}, ` + + `graph=${graphExecutionModule.publicSchema}.${graphExecutionModule.nodeStatesTable}` + ); + + return config; + } + + invalidate(databaseId: string): void { + this.cache.delete(databaseId); + } + + invalidateAll(): void { + this.cache.clear(); + } +} diff --git a/packages/module-loader/src/index.ts b/packages/module-loader/src/index.ts new file mode 100644 index 000000000..307645487 --- /dev/null +++ b/packages/module-loader/src/index.ts @@ -0,0 +1,41 @@ +/** + * @constructive-io/module-loader + * + * Unified MetaSchema module loader for the Constructive platform. + * Resolves compute, usage, and billing table/schema names dynamically + * with TTL caching. Multi-database (platform + tenant) aware. + * + * Usage: + * import { ModuleLoader, getModuleLoader } from '@constructive-io/module-loader'; + * + * const loader = getModuleLoader(pool); + * const cfg = await loader.compute(); // platform scope + * const cfg = await loader.compute(tenantId); // tenant scope + * loader.logCompute(entry); // fire-and-forget + */ + +// Core +export { TtlCache } from './cache'; +export { ModuleLoader, getModuleLoader, _resetModuleLoaderCache } from './module-loader'; + +// Domain loaders +export { ComputeModuleLoader } from './compute-loader'; +export type { ComputeModuleConfigExtended } from './compute-loader'; +export { UsageLoader, USAGE_DEFAULTS } from './usage-loader'; +export { BillingLoader } from './billing-loader'; + +// Types +export type { + BillingModuleConfig, + ComputeLogModuleConfig, + ComputeModuleConfig, + FunctionModuleConfig, + GraphExecutionModuleConfig, + InferenceEntry, + InvocationModuleConfig, + MeterEntry, + ModuleLoaderOptions, + StorageEntry, + UsageTableConfig, +} from './types'; +export { DEFAULT_DATABASE_ID, DEFAULT_TTL_MS } from './types'; diff --git a/packages/module-loader/src/module-loader.ts b/packages/module-loader/src/module-loader.ts new file mode 100644 index 000000000..b5c086cd7 --- /dev/null +++ b/packages/module-loader/src/module-loader.ts @@ -0,0 +1,132 @@ +/** + * ModuleLoader — unified entry point for all MetaSchema module resolution. + * + * Provides a single instance per pool that lazily resolves: + * - Compute modules (function definitions, invocations, compute logs, graph execution) + * - Usage modules (metering table names) + * - Billing modules (quota + usage recording) + * + * Multi-database (platform + tenant): + * const loader = new ModuleLoader({ pool }); + * loader.compute(); // platform scope (default) + * loader.compute(tenantDbId); // tenant scope + * loader.usage(tenantDbId); // tenant usage tables + * + * Scope-aware metering: + * loader.logCompute(entry); // resolves scope from entry.databaseId + * loader.logInference(entry); + * loader.logStorage(entry); + */ + +import type { Pool } from 'pg'; + +import { BillingLoader } from './billing-loader'; +import { ComputeModuleLoader } from './compute-loader'; +import type { ComputeModuleConfigExtended } from './compute-loader'; +import type { + BillingModuleConfig, + InferenceEntry, + MeterEntry, + ModuleLoaderOptions, + StorageEntry, + UsageTableConfig, +} from './types'; +import { DEFAULT_DATABASE_ID, DEFAULT_TTL_MS } from './types'; +import { UsageLoader } from './usage-loader'; + +export class ModuleLoader { + readonly computeLoader: ComputeModuleLoader; + readonly usageLoader: UsageLoader; + readonly billingLoader: BillingLoader; + + private pool: Pool; + private defaultDatabaseId: string; + + constructor(opts: ModuleLoaderOptions) { + this.pool = opts.pool; + this.defaultDatabaseId = opts.databaseId ?? DEFAULT_DATABASE_ID; + const ttl = opts.cacheTtlMs ?? DEFAULT_TTL_MS; + + this.computeLoader = new ComputeModuleLoader(this.pool, ttl); + this.usageLoader = new UsageLoader(this.pool, this.defaultDatabaseId, ttl); + this.billingLoader = new BillingLoader(this.pool, this.defaultDatabaseId, ttl); + } + + // ─── Compute Resolution ───────────────────────────────────────────────── + + compute(databaseId?: string): Promise { + return this.computeLoader.load(databaseId ?? this.defaultDatabaseId); + } + + // ─── Usage Resolution ─────────────────────────────────────────────────── + + usage(databaseId?: string): Promise { + return this.usageLoader.resolve(databaseId); + } + + // ─── Billing Resolution ───────────────────────────────────────────────── + + billing(databaseId?: string): Promise { + return this.billingLoader.load(databaseId); + } + + // ─── Fire-and-forget Metering ─────────────────────────────────────────── + + logCompute(entry: MeterEntry): string { + return this.usageLoader.logComputeUsage(entry); + } + + logInference(entry: InferenceEntry): void { + this.usageLoader.logInferenceUsage(entry); + } + + logStorage(entry: StorageEntry): void { + this.usageLoader.logStorageUsage(entry); + } + + // ─── Billing Convenience ──────────────────────────────────────────────── + + checkQuota(entityId: string, meterSlug: string, amount?: number, databaseId?: string): Promise { + return this.billingLoader.checkQuota(entityId, meterSlug, amount, databaseId); + } + + recordUsage(entityId: string, meterSlug: string, amount: number, metadata: Record, databaseId?: string): Promise { + return this.billingLoader.recordUsage(entityId, meterSlug, amount, metadata, databaseId); + } + + // ─── Cache Control ────────────────────────────────────────────────────── + + invalidate(databaseId?: string): void { + if (databaseId) { + this.computeLoader.invalidate(databaseId); + this.usageLoader.invalidate(databaseId); + this.billingLoader.invalidate(databaseId); + } else { + this.computeLoader.invalidateAll(); + this.usageLoader.invalidate(); + this.billingLoader.invalidate(); + } + } +} + +// ─── Convenience Factory ────────────────────────────────────────────────────── + +let _instance: ModuleLoader | null = null; +let _pool: Pool | null = null; + +/** + * Get or create a ModuleLoader for the given pool. + * Reuses the same instance as long as the pool reference matches. + */ +export function getModuleLoader(pool: Pool, databaseId?: string): ModuleLoader { + if (_instance && _pool === pool) return _instance; + _instance = new ModuleLoader({ pool, databaseId }); + _pool = pool; + return _instance; +} + +/** Reset the module-level cache (for testing). */ +export function _resetModuleLoaderCache(): void { + _instance = null; + _pool = null; +} diff --git a/packages/module-loader/src/types.ts b/packages/module-loader/src/types.ts new file mode 100644 index 000000000..3e209ff2a --- /dev/null +++ b/packages/module-loader/src/types.ts @@ -0,0 +1,127 @@ +/** + * Shared types for all module loaders. + */ + +import type { Pool } from 'pg'; + +// ─── Compute Module ────────────────────────────────────────────────────────── + +export interface FunctionModuleConfig { + publicSchema: string; + privateSchema: string; + definitionsTable: string; + secretDefinitionsTable: string; + scope: string; +} + +export interface InvocationModuleConfig { + publicSchema: string; + invocationsTable: string; + executionLogsTable: string; + scope: string; +} + +export interface ComputeLogModuleConfig { + publicSchema: string; + privateSchema: string; + computeLogTable: string; + usageDailyTable: string; + scope: string; +} + +export interface ComputeModuleConfig { + functionModule: FunctionModuleConfig | null; + invocationModules: InvocationModuleConfig[]; + computeLogModule: ComputeLogModuleConfig | null; +} + +// ─── Usage Module ──────────────────────────────────────────────────────────── + +export interface UsageTableConfig { + invocationsSchema: string; + invocationsTable: string; + computeUsageSchema: string; + computeUsageTable: string; + inferenceUsageSchema: string; + inferenceUsageTable: string; + storageUsageSchema: string; + storageUsageTable: string; +} + +// ─── Billing Module ────────────────────────────────────────────────────────── + +export interface BillingModuleConfig { + publicSchema: string; + privateSchema: string; + recordUsageFunction: string; +} + +// ─── Graph Execution Module ────────────────────────────────────────────────── + +export interface GraphExecutionModuleConfig { + publicSchema: string; + privateSchema: string; + nodeStatesTable: string; + completeNodeFunction: string; + failNodeFunction: string; +} + +// ─── Metering Entry Types ──────────────────────────────────────────────────── + +export interface MeterEntry { + jobId: number | string; + taskIdentifier: string; + databaseId?: string; + actorId?: string; + entityId?: string; + durationMs: number; + status: 'ok' | 'error'; + error?: string; + payload?: unknown; + result?: unknown; + graphExecutionId?: string; + nodeName?: string; + dispatchType: 'inline' | 'http'; +} + +export interface InferenceEntry { + databaseId?: string; + entityId?: string; + actorId?: string; + requestId?: string; + model: string; + provider: string; + service: 'chat' | 'embed'; + operation: string; + inputTokens: number; + outputTokens: number; + totalTokens: number; + latencyMs: number; + status: 'ok' | 'error'; + errorType?: string; + rawUsage?: unknown; +} + +export interface StorageEntry { + databaseId?: string; + entityId?: string; + actorId?: string; + operation: 'read' | 'write' | 'delete'; + bucket: string; + key: string; + sizeBytes: number; + durationMs: number; +} + +// ─── Loader Options ────────────────────────────────────────────────────────── + +export interface ModuleLoaderOptions { + pool: Pool; + /** Default database_id for lookups (platform = 00000000-...) */ + databaseId?: string; + /** Cache TTL in ms (default 60s) */ + cacheTtlMs?: number; +} + +export const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000'; +export const DEFAULT_TTL_MS = 60_000; diff --git a/packages/module-loader/src/usage-loader.ts b/packages/module-loader/src/usage-loader.ts new file mode 100644 index 000000000..4cb3219c1 --- /dev/null +++ b/packages/module-loader/src/usage-loader.ts @@ -0,0 +1,279 @@ +/** + * UsageLoader — resolves usage/metering table names from MetaSchema + * and provides fire-and-forget INSERT helpers. + * + * Multi-database: pass different databaseId to resolve tables for + * platform scope vs tenant scope. + */ + +import { Logger } from '@pgpmjs/logger'; +import { randomUUID } from 'crypto'; +import type { Pool } from 'pg'; + +import { TtlCache } from './cache'; +import type { InferenceEntry, MeterEntry, StorageEntry, UsageTableConfig } from './types'; +import { DEFAULT_DATABASE_ID, DEFAULT_TTL_MS } from './types'; + +const log = new Logger('module-loader:usage'); + +// ─── MetaSchema Resolution ──────────────────────────────────────────────────── + +const USAGE_TABLES_SQL = ` + SELECT s.schema_name, t.name AS table_name + FROM metaschema_public."table" t + JOIN metaschema_public.schema s ON t.schema_id = s.id + WHERE t.database_id = $1 + AND t.name = ANY($2) +`; + +const KNOWN_TABLES = [ + 'platform_function_invocations', + 'platform_usage_log_computes', + 'platform_usage_log_inferences', + 'platform_usage_log_storage', +] as const; + +/** Default config used when MetaSchema is unavailable */ +export const USAGE_DEFAULTS: UsageTableConfig = { + invocationsSchema: 'constructive_compute_public', + invocationsTable: 'platform_function_invocations', + computeUsageSchema: 'constructive_usage_public', + computeUsageTable: 'platform_usage_log_computes', + inferenceUsageSchema: 'constructive_usage_public', + inferenceUsageTable: 'platform_usage_log_inferences', + storageUsageSchema: 'constructive_usage_public', + storageUsageTable: 'platform_usage_log_storage', +}; + +// ─── UsageLoader Class ──────────────────────────────────────────────────────── + +export class UsageLoader { + private pool: Pool; + private defaultDatabaseId: string; + private cache: TtlCache; + + constructor(pool: Pool, databaseId: string = DEFAULT_DATABASE_ID, ttlMs: number = DEFAULT_TTL_MS) { + this.pool = pool; + this.defaultDatabaseId = databaseId; + this.cache = new TtlCache(ttlMs); + } + + /** + * Resolve usage table config for a database. + * Multi-database: call with different databaseId for platform vs tenant. + */ + async resolve(databaseId?: string): Promise { + const dbId = databaseId ?? this.defaultDatabaseId; + const cached = this.cache.get(dbId); + if (cached !== undefined) return cached; + + try { + const { rows } = await this.pool.query(USAGE_TABLES_SQL, [ + dbId, + [...KNOWN_TABLES], + ]); + + const config: UsageTableConfig = { ...USAGE_DEFAULTS }; + for (const row of rows) { + switch (row.table_name) { + case 'platform_function_invocations': + config.invocationsSchema = row.schema_name; + break; + case 'platform_usage_log_computes': + config.computeUsageSchema = row.schema_name; + break; + case 'platform_usage_log_inferences': + config.inferenceUsageSchema = row.schema_name; + break; + case 'platform_usage_log_storage': + config.storageUsageSchema = row.schema_name; + break; + } + } + + this.cache.set(dbId, config); + return config; + } catch { + log.debug(`metaschema lookup unavailable for database ${dbId} — using defaults`); + this.cache.set(dbId, USAGE_DEFAULTS); + return USAGE_DEFAULTS; + } + } + + // ─── Compute Metering ───────────────────────────────────────────────── + + /** + * Log a compute invocation. Fire-and-forget: never throws. + * Returns the generated invocation_id (UUID). + * + * Multi-database: uses entry.databaseId to resolve the correct tenant's tables. + */ + logComputeUsage(entry: MeterEntry): string { + const invocationId = randomUUID(); + this.resolve(entry.databaseId) + .then((cfg) => this.insertCompute(cfg, entry, invocationId)) + .catch((err) => { + log.warn(`compute usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + return invocationId; + } + + private async insertCompute(cfg: UsageTableConfig, entry: MeterEntry, invocationId: string): Promise { + const now = new Date(); + const startedAt = new Date(now.getTime() - Math.round(entry.durationMs)); + + this.pool + .query( + `INSERT INTO "${cfg.invocationsSchema}"."${cfg.invocationsTable}" + (id, database_id, actor_id, task_identifier, job_id, + graph_execution_id, status, duration_ms, + started_at, completed_at, payload, result, error, + entity_id, node_name, dispatch_type) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)`, + [ + invocationId, + entry.databaseId ?? null, + entry.actorId ?? null, + entry.taskIdentifier, + typeof entry.jobId === 'string' ? parseInt(entry.jobId, 10) || 0 : entry.jobId, + entry.graphExecutionId ?? null, + entry.status, + Math.round(entry.durationMs), + startedAt, + now, + entry.payload ? JSON.stringify(entry.payload) : null, + entry.result ? JSON.stringify(entry.result) : null, + entry.error ?? null, + entry.entityId ?? null, + entry.nodeName ?? null, + entry.dispatchType, + ] + ) + .catch((err) => { + log.warn(`invocation log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + + this.pool + .query( + `INSERT INTO "${cfg.computeUsageSchema}"."${cfg.computeUsageTable}" + (id, database_id, entity_id, actor_id, task_identifier, + job_id, invocation_id, status, duration_ms, error, completed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`, + [ + randomUUID(), + entry.databaseId ?? null, + entry.entityId ?? null, + entry.actorId ?? null, + entry.taskIdentifier, + typeof entry.jobId === 'string' ? parseInt(entry.jobId, 10) || 0 : entry.jobId, + invocationId, + entry.status, + Math.round(entry.durationMs), + entry.error ?? null, + now, + ] + ) + .catch((err) => { + log.warn(`compute usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + } + + // ─── Inference Metering ─────────────────────────────────────────────── + + /** + * Log an inference invocation. Fire-and-forget: never throws. + */ + logInferenceUsage(entry: InferenceEntry): void { + this.resolve(entry.databaseId) + .then((cfg) => this.insertInference(cfg, entry)) + .catch((err) => { + log.warn(`inference usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + } + + private async insertInference(cfg: UsageTableConfig, entry: InferenceEntry): Promise { + const id = randomUUID(); + const now = new Date(); + + this.pool + .query( + `INSERT INTO "${cfg.inferenceUsageSchema}"."${cfg.inferenceUsageTable}" + (id, database_id, entity_id, actor_id, request_id, + model, provider, service, operation, + input_tokens, output_tokens, total_tokens, + latency_ms, status, error_type, raw_usage, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`, + [ + id, + entry.databaseId ?? null, + entry.entityId ?? null, + entry.actorId ?? null, + entry.requestId ?? randomUUID(), + entry.model, + entry.provider, + entry.service, + entry.operation, + entry.inputTokens, + entry.outputTokens, + entry.totalTokens, + Math.round(entry.latencyMs), + entry.status, + entry.errorType ?? null, + entry.rawUsage ? JSON.stringify(entry.rawUsage) : null, + now, + ] + ) + .catch((err) => { + log.warn(`inference log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + } + + // ─── Storage Metering ───────────────────────────────────────────────── + + /** + * Log a storage operation. Fire-and-forget: never throws. + */ + logStorageUsage(entry: StorageEntry): void { + this.resolve(entry.databaseId) + .then((cfg) => this.insertStorage(cfg, entry)) + .catch((err) => { + log.warn(`storage usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + } + + private async insertStorage(cfg: UsageTableConfig, entry: StorageEntry): Promise { + const id = randomUUID(); + const now = new Date(); + + this.pool + .query( + `INSERT INTO "${cfg.storageUsageSchema}"."${cfg.storageUsageTable}" + (id, database_id, entity_id, actor_id, operation, + bucket, key, size_bytes, duration_ms, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, + [ + id, + entry.databaseId ?? null, + entry.entityId ?? null, + entry.actorId ?? null, + entry.operation, + entry.bucket, + entry.key, + entry.sizeBytes, + Math.round(entry.durationMs), + now, + ] + ) + .catch((err) => { + log.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`); + }); + } + + invalidate(databaseId?: string): void { + if (databaseId) { + this.cache.delete(databaseId); + } else { + this.cache.clear(); + } + } +} diff --git a/packages/module-loader/tsconfig.json b/packages/module-loader/tsconfig.json new file mode 100644 index 000000000..a45c52754 --- /dev/null +++ b/packages/module-loader/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "declaration": true + }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "node_modules"] +} diff --git a/packages/usage-loader/package.json b/packages/usage-loader/package.json new file mode 100644 index 000000000..f9716fc4a --- /dev/null +++ b/packages/usage-loader/package.json @@ -0,0 +1,26 @@ +{ + "name": "@constructive-io/usage-loader", + "version": "1.0.0", + "description": "Shared usage metering module loader — resolves table names from MetaSchema with TTL caching.", + "author": "Constructive ", + "license": "SEE LICENSE IN LICENSE", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "files": [ + "dist", + "README.md" + ], + "publishConfig": { + "access": "public" + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "clean": "rimraf dist" + }, + "dependencies": { + "@constructive-io/module-loader": "workspace:^" + }, + "peerDependencies": { + "pg": "^8.0.0" + } +} diff --git a/packages/usage-loader/src/index.ts b/packages/usage-loader/src/index.ts new file mode 100644 index 000000000..2c89c34de --- /dev/null +++ b/packages/usage-loader/src/index.ts @@ -0,0 +1,21 @@ +/** + * @constructive-io/usage-loader + * + * Backward-compatibility re-export from @constructive-io/module-loader. + * The canonical implementation now lives in packages/module-loader. + */ + +export { + UsageLoader, + USAGE_DEFAULTS as DEFAULTS, + TtlCache, + getModuleLoader as getLoader, + _resetModuleLoaderCache as _resetLoaderCache, +} from '@constructive-io/module-loader'; + +export type { + UsageTableConfig, + MeterEntry, + InferenceEntry, + StorageEntry, +} from '@constructive-io/module-loader'; diff --git a/packages/usage-loader/tsconfig.json b/packages/usage-loader/tsconfig.json new file mode 100644 index 000000000..a45c52754 --- /dev/null +++ b/packages/usage-loader/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "declaration": true + }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "node_modules"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6485c478f..055e452c7 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -211,6 +211,9 @@ importers: '@constructive-io/job-utils': specifier: ^2.5.4 version: 2.5.4 + '@constructive-io/module-loader': + specifier: workspace:^ + version: link:../../packages/module-loader '@pgpmjs/logger': specifier: ^2.4.3 version: 2.5.2 @@ -331,6 +334,12 @@ importers: '@constructive-io/job-utils': specifier: ^2.5.4 version: 2.5.4 + '@constructive-io/module-loader': + specifier: workspace:^ + version: link:../../packages/module-loader + '@constructive-io/usage-loader': + specifier: workspace:^ + version: link:../../packages/usage-loader '@pgpmjs/logger': specifier: ^2.4.3 version: 2.5.2 @@ -353,6 +362,9 @@ importers: packages/agentic-server: dependencies: + '@constructive-io/usage-loader': + specifier: workspace:^ + version: link:../usage-loader '@pgpmjs/logger': specifier: ^2.4.3 version: 2.12.0 @@ -546,6 +558,9 @@ importers: '@constructive-io/knative-job-fn': specifier: workspace:^ version: link:../fn-app + '@constructive-io/usage-loader': + specifier: workspace:^ + version: link:../usage-loader '@pgpmjs/logger': specifier: ^2.4.3 version: 2.5.2 @@ -607,6 +622,24 @@ importers: specifier: ^5.8.3 version: 5.9.3 + packages/module-loader: + dependencies: + '@pgpmjs/logger': + specifier: ^2.4.3 + version: 2.12.0 + pg: + specifier: ^8.0.0 + version: 8.21.0 + + packages/usage-loader: + dependencies: + '@constructive-io/module-loader': + specifier: workspace:^ + version: link:../module-loader + pg: + specifier: ^8.0.0 + version: 8.21.0 + sdk/constructive-functions-cli: dependencies: '@0no-co/graphql.web': diff --git a/tests/integration/inline-nodes.test.ts b/tests/integration/inline-nodes.test.ts index db77278b3..558cb88c7 100644 --- a/tests/integration/inline-nodes.test.ts +++ b/tests/integration/inline-nodes.test.ts @@ -13,7 +13,7 @@ import { isGraphJob, extractNodeProps } from '../../job/worker/src/inline-nodes'; -import { completeNode, failNode } from '../../job/worker/src/graph-complete'; +import { completeNode, failNode, _resetGraphCompleteCache } from '../../job/worker/src/graph-complete'; // --------------------------------------------------------------------------- // 1. Inline node registry @@ -215,6 +215,8 @@ describe('extractNodeProps', () => { // --------------------------------------------------------------------------- describe('completeNode', () => { + beforeEach(() => _resetGraphCompleteCache()); + it('calls platform_complete_node SQL with correct args', async () => { const mockQuery = jest.fn().mockResolvedValue({ rows: [] }); const mockPool = { query: mockQuery } as any; @@ -226,8 +228,9 @@ describe('completeNode', () => { { sum: 10 } ); - expect(mockQuery).toHaveBeenCalledTimes(1); - const [sql, params] = mockQuery.mock.calls[0]; + // Last call is the actual complete_node; earlier calls are MetaSchema resolution + const lastCall = mockQuery.mock.calls[mockQuery.mock.calls.length - 1]; + const [sql, params] = lastCall; expect(sql).toContain('platform_complete_node'); expect(params).toEqual([ 'exec-uuid-123', @@ -238,6 +241,8 @@ describe('completeNode', () => { }); describe('failNode', () => { + beforeEach(() => _resetGraphCompleteCache()); + it('calls platform_fail_node SQL with correct args', async () => { const mockQuery = jest.fn().mockResolvedValue({ rows: [] }); const mockPool = { query: mockQuery } as any; @@ -249,8 +254,9 @@ describe('failNode', () => { 'impl threw an error' ); - expect(mockQuery).toHaveBeenCalledTimes(1); - const [sql, params] = mockQuery.mock.calls[0]; + // Last call is the actual fail_node; earlier calls are MetaSchema resolution + const lastCall = mockQuery.mock.calls[mockQuery.mock.calls.length - 1]; + const [sql, params] = lastCall; expect(sql).toContain('platform_fail_node'); expect(params).toEqual([ 'exec-uuid-123', diff --git a/tests/integration/multi-provider.test.ts b/tests/integration/multi-provider.test.ts new file mode 100644 index 000000000..0d3b713ca --- /dev/null +++ b/tests/integration/multi-provider.test.ts @@ -0,0 +1,557 @@ +/** + * Integration tests for multi-provider routing and /v1/usage endpoint. + * + * Verifies: + * 1. Multi-provider routing by model name prefix + * 2. Multi-provider routing by X-Provider header + * 3. Model-based auto-routing (claude-* → anthropic, gpt-* → openai) + * 4. /v1/usage endpoint accepts external usage reports + * 5. /v1/providers lists configured providers + * 6. Anthropic response transformation to OpenAI format + */ + +import express from 'express'; +import http from 'http'; +import { createAgenticServer } from '../../packages/agentic-server/src/server'; +import { _resetCache } from '../../packages/agentic-server/src/inference-meter'; + +const flush = () => new Promise((r) => setTimeout(r, 30)); + +// ── Mock Providers ──────────────────────────────────────────────────────────── + +function createMockOpenAI() { + const app = express(); + app.use(express.json()); + const calls: any[] = []; + + app.post('/v1/chat/completions', (req: any, res: any) => { + calls.push({ provider: 'openai', body: req.body }); + res.json({ + id: 'chatcmpl-openai', + object: 'chat.completion', + choices: [{ message: { role: 'assistant', content: 'Hello from OpenAI!' }, finish_reason: 'stop', index: 0 }], + usage: { prompt_tokens: 20, completion_tokens: 10, total_tokens: 30 } + }); + }); + + app.post('/v1/embeddings', (req: any, res: any) => { + calls.push({ provider: 'openai', path: '/v1/embeddings', body: req.body }); + res.json({ + object: 'list', + data: [{ object: 'embedding', embedding: [0.1, 0.2, 0.3], index: 0 }], + usage: { prompt_tokens: 5, total_tokens: 5 } + }); + }); + + return { app, calls }; +} + +function createMockAnthropic() { + const app = express(); + app.use(express.json()); + const calls: any[] = []; + + app.post('/v1/messages', (req: any, res: any) => { + calls.push({ provider: 'anthropic', body: req.body, headers: req.headers }); + res.json({ + id: 'msg_anthropic', + type: 'message', + role: 'assistant', + content: [{ type: 'text', text: 'Hello from Anthropic!' }], + usage: { input_tokens: 15, output_tokens: 8 } + }); + }); + + return { app, calls }; +} + +function createMockOllama() { + const app = express(); + app.use(express.json()); + const calls: any[] = []; + + app.post('/api/chat', (req: any, res: any) => { + calls.push({ provider: 'ollama', body: req.body }); + res.json({ + message: { role: 'assistant', content: 'Hello from Ollama!' }, + done: true, + prompt_eval_count: 12, + eval_count: 6 + }); + }); + + app.post('/api/embed', (req: any, res: any) => { + calls.push({ provider: 'ollama', path: '/api/embed', body: req.body }); + res.json({ + model: req.body.model, + embeddings: [[0.4, 0.5, 0.6]] + }); + }); + + return { app, calls }; +} + +// ── Test Suite ──────────────────────────────────────────────────────────────── + +describe('Multi-provider routing', () => { + let openai: ReturnType; + let anthropic: ReturnType; + let ollama: ReturnType; + let openaiServer: http.Server; + let anthropicServer: http.Server; + let ollamaServer: http.Server; + let openaiPort: number; + let anthropicPort: number; + let ollamaPort: number; + let mockQuery: jest.Mock; + let mockPool: any; + + beforeAll(async () => { + openai = createMockOpenAI(); + anthropic = createMockAnthropic(); + ollama = createMockOllama(); + + [openaiServer, anthropicServer, ollamaServer] = await Promise.all([ + new Promise((resolve) => { const s = openai.app.listen(0, () => resolve(s)); }), + new Promise((resolve) => { const s = anthropic.app.listen(0, () => resolve(s)); }), + new Promise((resolve) => { const s = ollama.app.listen(0, () => resolve(s)); }) + ]); + + openaiPort = (openaiServer.address() as { port: number }).port; + anthropicPort = (anthropicServer.address() as { port: number }).port; + ollamaPort = (ollamaServer.address() as { port: number }).port; + }); + + afterAll(async () => { + await Promise.all([ + new Promise((r) => openaiServer.close(() => r())), + new Promise((r) => anthropicServer.close(() => r())), + new Promise((r) => ollamaServer.close(() => r())) + ]); + }); + + beforeEach(() => { + _resetCache(); + mockQuery = jest.fn().mockResolvedValue({ rows: [] }); + mockPool = { query: mockQuery } as any; + openai.calls.length = 0; + anthropic.calls.length = 0; + ollama.calls.length = 0; + }); + + function createMultiServer() { + return createAgenticServer({ + providers: [ + { type: 'openai', baseUrl: `http://127.0.0.1:${openaiPort}`, apiKey: 'sk-test-openai', defaultModel: 'gpt-4o' }, + { type: 'anthropic', baseUrl: `http://127.0.0.1:${anthropicPort}`, apiKey: 'sk-ant-test', defaultModel: 'claude-sonnet-4-20250514' }, + { type: 'ollama', baseUrl: `http://127.0.0.1:${ollamaPort}`, defaultModel: 'llama3' } + ], + pgPool: mockPool + }); + } + + async function withServer(app: any, fn: (port: number) => Promise) { + const server = await new Promise((resolve) => { + const s = app.listen(0, () => resolve(s)); + }); + try { + await fn((server.address() as { port: number }).port); + } finally { + await new Promise((r) => server.close(() => r())); + } + } + + // ─── Routing Tests ────────────────────────────────────────────────────── + + it('routes gpt-* models to OpenAI provider', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'gpt-4o', messages: [{ role: 'user', content: 'Hi' }] }) + }); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.choices[0].message.content).toBe('Hello from OpenAI!'); + expect(openai.calls).toHaveLength(1); + expect(anthropic.calls).toHaveLength(0); + expect(ollama.calls).toHaveLength(0); + }); + }); + + it('routes claude-* models to Anthropic provider', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'claude-sonnet-4-20250514', messages: [{ role: 'user', content: 'Hi' }] }) + }); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.choices[0].message.content).toBe('Hello from Anthropic!'); + expect(anthropic.calls).toHaveLength(1); + expect(openai.calls).toHaveLength(0); + }); + }); + + it('routes llama-* models to Ollama provider', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'llama3', messages: [{ role: 'user', content: 'Hi' }] }) + }); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.choices[0].message.content).toBe('Hello from Ollama!'); + expect(ollama.calls).toHaveLength(1); + expect(openai.calls).toHaveLength(0); + }); + }); + + it('routes by X-Provider header (overrides model name)', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'test', + 'X-Provider': 'ollama' + }, + body: JSON.stringify({ model: 'gpt-4o', messages: [{ role: 'user', content: 'Hi' }] }) + }); + expect(res.status).toBe(200); + // Even though model is gpt-4o, X-Provider forces ollama + expect(ollama.calls).toHaveLength(1); + expect(openai.calls).toHaveLength(0); + }); + }); + + it('routes by model prefix notation: "anthropic/claude-3.5-sonnet"', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'anthropic/claude-3.5-sonnet', messages: [{ role: 'user', content: 'Hi' }] }) + }); + expect(res.status).toBe(200); + expect(anthropic.calls).toHaveLength(1); + // Model sent to Anthropic should strip the prefix + expect(anthropic.calls[0].body.model).toBe('claude-3.5-sonnet'); + }); + }); + + // ─── Anthropic Transform Tests ────────────────────────────────────────── + + it('transforms Anthropic response to OpenAI-compatible format', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'claude-sonnet-4-20250514', messages: [{ role: 'user', content: 'Hello' }] }) + }); + const body = await res.json() as any; + // Should be in OpenAI format + expect(body.object).toBe('chat.completion'); + expect(body.choices[0].message.role).toBe('assistant'); + expect(body.choices[0].message.content).toBe('Hello from Anthropic!'); + expect(body.usage.prompt_tokens).toBe(15); + expect(body.usage.completion_tokens).toBe(8); + expect(body.usage.total_tokens).toBe(23); + }); + }); + + it('sends x-api-key header for Anthropic (not Authorization Bearer)', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'claude-sonnet-4-20250514', messages: [{ role: 'user', content: 'Hi' }] }) + }); + const headers = anthropic.calls[0].headers; + expect(headers['x-api-key']).toBe('sk-ant-test'); + expect(headers['anthropic-version']).toBe('2023-06-01'); + expect(headers['authorization']).toBeUndefined(); + }); + }); + + it('extracts system message for Anthropic', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ + model: 'claude-sonnet-4-20250514', + messages: [ + { role: 'system', content: 'You are helpful.' }, + { role: 'user', content: 'Hi' } + ] + }) + }); + const body = anthropic.calls[0].body; + expect(body.system).toBe('You are helpful.'); + expect(body.messages).toHaveLength(1); + expect(body.messages[0].role).toBe('user'); + }); + }); + + // ─── Metering Tests ───────────────────────────────────────────────────── + + it('meters inference with correct provider name', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'test', + 'X-Database-Id': 'db-multi-001' + }, + body: JSON.stringify({ model: 'claude-sonnet-4-20250514', messages: [{ role: 'user', content: 'Hi' }] }) + }); + await flush(); + + const insertCalls = mockQuery.mock.calls.filter( + ([sql]: [string]) => sql.includes('INSERT INTO') && sql.includes('platform_usage_log_inferences') + ); + expect(insertCalls).toHaveLength(1); + const [, params] = insertCalls[0]; + expect(params[6]).toBe('anthropic'); // provider + expect(params[9]).toBe(15); // input_tokens + expect(params[10]).toBe(8); // output_tokens + expect(params[11]).toBe(23); // total_tokens + }); + }); + + // ─── Embeddings ───────────────────────────────────────────────────────── + + it('routes embedding requests to correct provider', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/embeddings`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Internal-Service': 'test' }, + body: JSON.stringify({ model: 'text-embedding-3-small', input: 'hello' }) + }); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.data[0].embedding).toEqual([0.1, 0.2, 0.3]); + // gpt/text-embedding → openai + expect(openai.calls).toHaveLength(1); + }); + }); + + it('routes ollama embeddings via X-Provider', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/embeddings`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'test', + 'X-Provider': 'ollama' + }, + body: JSON.stringify({ model: 'nomic-embed-text', input: 'hello' }) + }); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.data[0].embedding).toEqual([0.4, 0.5, 0.6]); + expect(ollama.calls).toHaveLength(1); + }); + }); + + // ─── /v1/providers Endpoint ───────────────────────────────────────────── + + it('lists configured providers', async () => { + const app = createMultiServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/providers`); + expect(res.status).toBe(200); + const body = await res.json() as any; + expect(body.providers).toHaveLength(3); + expect(body.providers.map((p: any) => p.type)).toEqual(['openai', 'anthropic', 'ollama']); + }); + }); +}); + +// ═══════════════════════════════════════════════════════════════════════════════ +// /v1/usage — External Usage Reporting +// ═══════════════════════════════════════════════════════════════════════════════ + +describe('/v1/usage endpoint', () => { + let mockQuery: jest.Mock; + let mockPool: any; + + beforeEach(() => { + _resetCache(); + mockQuery = jest.fn().mockResolvedValue({ rows: [] }); + mockPool = { query: mockQuery } as any; + }); + + function createServer() { + return createAgenticServer({ + providerBaseUrl: 'http://localhost:11434', + providerType: 'openai', + pgPool: mockPool + }); + } + + async function withServer(app: any, fn: (port: number) => Promise) { + const server = await new Promise((resolve) => { + const s = app.listen(0, () => resolve(s)); + }); + try { + await fn((server.address() as { port: number }).port); + } finally { + await new Promise((r) => server.close(() => r())); + } + } + + it('accepts external usage report and returns 202', async () => { + const app = createServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/usage`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'fn-runtime', + 'X-Database-Id': 'db-python-001', + 'X-Entity-Id': 'entity-py', + 'X-Actor-Id': 'actor-py' + }, + body: JSON.stringify({ + model: 'llama3-8b', + provider: 'llamaparse', + service: 'chat', + operation: 'local-inference', + input_tokens: 500, + output_tokens: 200, + total_tokens: 700, + latency_ms: 1200, + status: 'ok' + }) + }); + expect(res.status).toBe(202); + const body = await res.json() as any; + expect(body.accepted).toBe(true); + }); + }); + + it('fires inference metering INSERT with reported values', async () => { + const app = createServer(); + await withServer(app, async (port) => { + await fetch(`http://127.0.0.1:${port}/v1/usage`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'fn-runtime', + 'X-Database-Id': 'db-python-002', + 'X-Entity-Id': 'entity-usage', + 'X-Actor-Id': 'actor-usage' + }, + body: JSON.stringify({ + model: 'huggingface/meta-llama/Llama-3-8B', + provider: 'huggingface', + service: 'chat', + operation: 'text-generation', + input_tokens: 1000, + output_tokens: 500, + total_tokens: 1500, + latency_ms: 3500, + status: 'ok', + raw_usage: { custom_field: 'metadata' } + }) + }); + await flush(); + + const insertCalls = mockQuery.mock.calls.filter( + ([sql]: [string]) => sql.includes('INSERT INTO') && sql.includes('platform_usage_log_inferences') + ); + expect(insertCalls).toHaveLength(1); + const [, params] = insertCalls[0]; + expect(params[1]).toBe('db-python-002'); // database_id + expect(params[2]).toBe('entity-usage'); // entity_id + expect(params[3]).toBe('actor-usage'); // actor_id + expect(params[5]).toBe('huggingface/meta-llama/Llama-3-8B'); // model + expect(params[6]).toBe('huggingface'); // provider + expect(params[7]).toBe('chat'); // service + expect(params[8]).toBe('text-generation'); // operation + expect(params[9]).toBe(1000); // input_tokens + expect(params[10]).toBe(500); // output_tokens + expect(params[11]).toBe(1500); // total_tokens + expect(params[12]).toBe(3500); // latency_ms + expect(params[13]).toBe('ok'); // status + }); + }); + + it('returns 400 when model is missing', async () => { + const app = createServer(); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/usage`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ provider: 'local', input_tokens: 100 }) + }); + expect(res.status).toBe(400); + const body = await res.json() as any; + expect(body.error.message).toContain('model is required'); + }); + }); + + it('handles error status reports', async () => { + const app = createServer(); + await withServer(app, async (port) => { + await fetch(`http://127.0.0.1:${port}/v1/usage`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Internal-Service': 'fn-runtime', + 'X-Database-Id': 'db-err' + }, + body: JSON.stringify({ + model: 'gpt-4o', + provider: 'openai', + status: 'error', + error_type: 'rate_limited', + input_tokens: 0, + output_tokens: 0, + total_tokens: 0, + latency_ms: 50 + }) + }); + await flush(); + + const insertCalls = mockQuery.mock.calls.filter( + ([sql]: [string]) => sql.includes('INSERT INTO') && sql.includes('platform_usage_log_inferences') + ); + expect(insertCalls).toHaveLength(1); + const [, params] = insertCalls[0]; + expect(params[13]).toBe('error'); // status + expect(params[14]).toBe('rate_limited'); // error_type + }); + }); + + it('works without pgPool configured (graceful no-op)', async () => { + const app = createAgenticServer({ + providerBaseUrl: 'http://localhost:11434', + providerType: 'openai' + // no pgPool + }); + await withServer(app, async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/v1/usage`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'local-model', input_tokens: 10 }) + }); + expect(res.status).toBe(202); + }); + }); +}); diff --git a/tests/integration/storage-meter.test.ts b/tests/integration/storage-meter.test.ts index 7bc5f40b1..7c9c98722 100644 --- a/tests/integration/storage-meter.test.ts +++ b/tests/integration/storage-meter.test.ts @@ -8,12 +8,20 @@ * 4. Handles null database_id, entity_id, actor_id gracefully * 5. Rounds duration_ms to nearest integer * 6. Works for read, write, and delete operations + * 7. Resolves table names via MetaSchema module loader */ import { logStorageUsage } from '../../job/worker/src/storage-meter'; /** Wait for fire-and-forget promises to settle */ -const flush = () => new Promise((r) => setTimeout(r, 20)); +const flush = () => new Promise((r) => setTimeout(r, 30)); + +/** Filter query calls to only INSERT statements for storage table */ +function storageInserts(mockQuery: jest.Mock) { + return mockQuery.mock.calls.filter( + ([sql]: [string]) => sql.includes('INSERT INTO') && sql.includes('platform_usage_log_storage') + ); +} describe('logStorageUsage', () => { let mockQuery: jest.Mock; @@ -24,7 +32,7 @@ describe('logStorageUsage', () => { mockPool = { query: mockQuery } as any; }); - it('inserts into platform_usage_log_storage', async () => { + it('inserts into platform_usage_log_storage via module loader', async () => { logStorageUsage(mockPool, { databaseId: 'db-001', entityId: 'entity-001', @@ -37,12 +45,30 @@ describe('logStorageUsage', () => { }); await flush(); - expect(mockQuery).toHaveBeenCalledTimes(1); - const [sql] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [sql] = inserts[0]; expect(sql).toContain('INSERT INTO'); expect(sql).toContain('platform_usage_log_storage'); }); + it('resolves table names from MetaSchema before INSERT', async () => { + logStorageUsage(mockPool, { + operation: 'read', + bucket: 'b', + key: 'k', + sizeBytes: 1, + durationMs: 1 + }); + await flush(); + + // First call is the MetaSchema lookup + const metaCalls = mockQuery.mock.calls.filter( + ([sql]: [string]) => sql.includes('metaschema_public') + ); + expect(metaCalls.length).toBeGreaterThanOrEqual(1); + }); + it('passes correct columns for a read operation', async () => { logStorageUsage(mockPool, { databaseId: 'db-aaa', @@ -56,7 +82,9 @@ describe('logStorageUsage', () => { }); await flush(); - const [sql, params] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [sql, params] = inserts[0]; expect(sql).toContain('database_id'); expect(sql).toContain('entity_id'); expect(sql).toContain('actor_id'); @@ -93,7 +121,9 @@ describe('logStorageUsage', () => { }); await flush(); - const [, params] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [, params] = inserts[0]; expect(params[4]).toBe('write'); // operation expect(params[5]).toBe('uploads'); // bucket expect(params[6]).toBe('docs/report.pdf'); // key @@ -112,7 +142,9 @@ describe('logStorageUsage', () => { }); await flush(); - const [, params] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [, params] = inserts[0]; expect(params[4]).toBe('delete'); // operation expect(params[5]).toBe('temp'); // bucket expect(params[6]).toBe('scratch/old-file.tmp'); // key @@ -129,9 +161,9 @@ describe('logStorageUsage', () => { }); await flush(); - expect(mockQuery).toHaveBeenCalledTimes(1); - - const [, params] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [, params] = inserts[0]; expect(params[1]).toBeNull(); // database_id expect(params[2]).toBeNull(); // entity_id expect(params[3]).toBeNull(); // actor_id @@ -151,7 +183,7 @@ describe('logStorageUsage', () => { }).not.toThrow(); await flush(); - expect(mockQuery).toHaveBeenCalledTimes(1); + // Both MetaSchema lookup and INSERT will fail, but no throw }); it('rounds duration_ms to nearest integer', async () => { @@ -164,7 +196,9 @@ describe('logStorageUsage', () => { }); await flush(); - const [, params] = mockQuery.mock.calls[0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(1); + const [, params] = inserts[0]; expect(params[8]).toBe(4); // Math.round(3.7) }); @@ -185,9 +219,10 @@ describe('logStorageUsage', () => { }); await flush(); - expect(mockQuery).toHaveBeenCalledTimes(2); - const id1 = mockQuery.mock.calls[0][1][0]; - const id2 = mockQuery.mock.calls[1][1][0]; + const inserts = storageInserts(mockQuery); + expect(inserts).toHaveLength(2); + const id1 = inserts[0][1][0]; + const id2 = inserts[1][1][0]; expect(id1).not.toBe(id2); }); });