feat: multi-provider routing + /v1/usage reporting endpoint

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"
   },

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<BillingModuleConfig | null>;
-  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<BillingModuleConfig | null>(cacheTtlMs ?? 60_000);
-  }
-
-  async load(databaseId?: string): Promise<BillingModuleConfig | null> {
-    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<boolean> {
-    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<string, unknown>,
-    databaseId?: string
-  ): Promise<void> {
-    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);
   }
 }

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<T> {
-  value: T;
-  expires_at: number;
-}
-
-export class TtlCache<T> {
-  private store = new Map<string, CacheEntry<T>>();
-  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';

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 {

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');

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<GraphExecutionModuleConfig> {
+    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<void> {
     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<void> {
     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<void> {
     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,
       });

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');